Events

Retrieve an event by id

EventsAPI.retrieve(id: Optional[int] = None, external_id: Optional[str] = None) cognite.client.data_classes.events.Event | None

Retrieve a single event by id.

Parameters
  • id (int | None) – ID

  • external_id (str | None) – External ID

Returns

Requested event or None if it does not exist.

Return type

Event | None

Examples

Get event by id:

>>> from cognite.client import CogniteClient
>>> client = CogniteClient()
>>> res = client.events.retrieve(id=1)

Get event by external id:

>>> res = client.events.retrieve(external_id="1")

Retrieve multiple events by id

EventsAPI.retrieve_multiple(ids: Optional[Sequence[int]] = None, external_ids: Optional[SequenceNotStr[str]] = None, ignore_unknown_ids: bool = False) EventList

Retrieve multiple events by id.

Parameters
  • ids (Sequence[int] | None) – IDs

  • external_ids (SequenceNotStr[str] | None) – External IDs

  • ignore_unknown_ids (bool) – Ignore IDs and external IDs that are not found rather than throw an exception.

Returns

The requested events.

Return type

EventList

Examples

Get events by id:

>>> from cognite.client import CogniteClient
>>> client = CogniteClient()
>>> res = client.events.retrieve_multiple(ids=[1, 2, 3])

Get events by external id:

>>> res = client.events.retrieve_multiple(external_ids=["abc", "def"])

List events

EventsAPI.list(start_time: Optional[Union[dict[str, Any], TimestampRange]] = None, end_time: Optional[Union[dict[str, Any], EndTimeFilter]] = None, active_at_time: Optional[Union[dict[str, Any], TimestampRange]] = None, type: Optional[str] = None, subtype: Optional[str] = None, metadata: Optional[dict[str, str]] = None, asset_ids: Optional[Sequence[int]] = None, asset_external_ids: Optional[SequenceNotStr[str]] = None, asset_subtree_ids: Optional[Union[int, Sequence[int]]] = None, asset_subtree_external_ids: Optional[Union[str, SequenceNotStr[str]]] = None, data_set_ids: Optional[Union[int, Sequence[int]]] = None, data_set_external_ids: Optional[Union[str, SequenceNotStr[str]]] = None, source: Optional[str] = None, created_time: Optional[Union[dict[str, Any], TimestampRange]] = None, last_updated_time: Optional[Union[dict[str, Any], TimestampRange]] = None, external_id_prefix: Optional[str] = None, sort: Optional[Union[EventSort, str, SortableEventProperty, tuple[str, Literal['asc', 'desc']], tuple[str, Literal['asc', 'desc'], Literal['auto', 'first', 'last']], list[cognite.client.data_classes.events.EventSort | str | cognite.client.data_classes.events.SortableEventProperty | tuple[str, Literal['asc', 'desc']] | tuple[str, Literal['asc', 'desc'], Literal['auto', 'first', 'last']]]]] = None, partitions: Optional[int] = None, limit: int | None = 25, advanced_filter: Optional[Union[Filter, dict[str, Any]]] = None) EventList

List events

Parameters
  • start_time (dict[str, Any] | TimestampRange | None) – Range between two timestamps.

  • end_time (dict[str, Any] | EndTimeFilter | None) – Range between two timestamps.

  • active_at_time (dict[str, Any] | TimestampRange | None) – Event is considered active from its startTime to endTime inclusive. If startTime is null, event is never active. If endTime is null, event is active from startTime onwards. activeAtTime filter will match all events that are active at some point from min to max, from min, or to max, depending on which of min and max parameters are specified.

  • type (str | None) – Type of the event, e.g ‘failure’.

  • subtype (str | None) – Subtype of the event, e.g ‘electrical’.

  • metadata (dict[str, str] | None) – Customizable extra data about the event. String key -> String value.

  • asset_ids (Sequence[int] | None) – Asset IDs of related equipments that this event relates to.

  • asset_external_ids (SequenceNotStr[str] | None) – Asset External IDs of related equipment that this event relates to.

  • asset_subtree_ids (int | Sequence[int] | None) – Only include events that have a related asset in a subtree rooted at any of these assetIds. If the total size of the given subtrees exceeds 100,000 assets, an error will be returned.

  • asset_subtree_external_ids (str | SequenceNotStr[str] | None) – Only include events that have a related asset in a subtree rooted at any of these assetExternalIds. If the total size of the given subtrees exceeds 100,000 assets, an error will be returned.

  • data_set_ids (int | Sequence[int] | None) – Return only events in the specified data set(s) with this id / these ids.

  • data_set_external_ids (str | SequenceNotStr[str] | None) – Return only events in the specified data set(s) with this external id / these external ids.

  • source (str | None) – The source of this event.

  • created_time (dict[str, Any] | TimestampRange | None) – Range between two timestamps. Possible keys are min and max, with values given as time stamps in ms.

  • last_updated_time (dict[str, Any] | TimestampRange | None) – Range between two timestamps. Possible keys are min and max, with values given as time stamps in ms.

  • external_id_prefix (str | None) – External Id provided by client. Should be unique within the project.

  • sort (SortSpec | list[SortSpec] | None) – The criteria to sort by. Defaults to desc for _score_ and asc for all other properties. Sort is not allowed if partitions is used.

  • partitions (int | None) – Retrieve resources in parallel using this number of workers (values up to 10 allowed), limit must be set to None (or -1).

  • limit (int | None) – Maximum number of events to return. Defaults to 25. Set to -1, float(“inf”) or None to return all items.

  • advanced_filter (Filter | dict[str, Any] | None) – Advanced filter query using the filter DSL (Domain Specific Language). It allows defining complex filtering expressions that combine simple operations, such as equals, prefix, exists, etc., using boolean operators and, or, and not. See examples below for usage.

Returns

List of requested events

Return type

EventList

Note

When using partitions, there are few considerations to keep in mind:
  • limit has to be set to None (or -1).

  • API may reject requests if you specify more than 10 partitions. When Cognite enforces this behavior, the requests result in a 400 Bad Request status.

  • Partitions are done independently of sorting: there’s no guarantee of the sort order between elements from different partitions. For this reason providing a sort parameter when using partitions is not allowed.

Examples

List events and filter on max start time:

>>> from cognite.client import CogniteClient
>>> client = CogniteClient()
>>> event_list = client.events.list(limit=5, start_time={"max": 1500000000})

Iterate over events:

>>> for event in client.events:
...     event # do something with the event

Iterate over chunks of events to reduce memory load:

>>> for event_list in client.events(chunk_size=2500):
...     event_list # do something with the events

Using advanced filter, find all events that have a metadata key ‘timezone’ starting with ‘Europe’, and sort by external id ascending:

>>> from cognite.client.data_classes import filters
>>> in_timezone = filters.Prefix(["metadata", "timezone"], "Europe")
>>> res = client.events.list(advanced_filter=in_timezone, sort=("external_id", "asc"))

Note that you can check the API documentation above to see which properties you can filter on with which filters.

To make it easier to avoid spelling mistakes and easier to look up available properties for filtering and sorting, you can also use the EventProperty and SortableEventProperty Enums.

>>> from cognite.client.data_classes import filters
>>> from cognite.client.data_classes.events import EventProperty, SortableEventProperty
>>> in_timezone = filters.Prefix(EventProperty.metadata_key("timezone"), "Europe")
>>> res = client.events.list(
...     advanced_filter=in_timezone,
...     sort=(SortableEventProperty.external_id, "asc"))

Combine filter and advanced filter:

>>> from cognite.client.data_classes import filters
>>> not_instrument_lvl5 = filters.And(
...    filters.ContainsAny("labels", ["Level5"]),
...    filters.Not(filters.ContainsAny("labels", ["Instrument"]))
... )
>>> res = client.events.list(asset_subtree_ids=[123456], advanced_filter=not_instrument_lvl5)

Aggregate events

EventsAPI.aggregate(filter: Optional[Union[EventFilter, dict[str, Any]]] = None) list[cognite.client.data_classes.shared.AggregateResult]

Aggregate events

Parameters

filter (EventFilter | dict[str, Any] | None) – Filter on events filter with exact match

Returns

List of event aggregates

Return type

list[AggregateResult]

Examples

Aggregate events:

>>> from cognite.client import CogniteClient
>>> client = CogniteClient()
>>> aggregate_type = client.events.aggregate(filter={"type": "failure"})

Aggregate Event Count

EventsAPI.aggregate_count(property: Optional[Union[EventProperty, str, list[str]]] = None, advanced_filter: Optional[Union[Filter, dict[str, Any]]] = None, filter: Optional[Union[EventFilter, dict[str, Any]]] = None) int

Count of event matching the specified filters.

Parameters
  • property (EventPropertyLike | None) – If specified, Get an approximate number of Events with a specific property (property is not null) and matching the filters.

  • advanced_filter (Filter | dict[str, Any] | None) – The filter to narrow down the events to count.

  • filter (EventFilter | dict[str, Any] | None) – The filter to narrow down the events to count requiring exact match.

Returns

The number of events matching the specified filters and search.

Return type

int

Examples

Count the number of events in your CDF project:

>>> from cognite.client import CogniteClient
>>> client = CogniteClient()
>>> count = client.events.aggregate_count()

Count the number of workorder events in your CDF project:

>>> from cognite.client.data_classes import filters
>>> from cognite.client.data_classes.events import EventProperty
>>> is_workorder = filters.Equals(EventProperty.type, "workorder")
>>> workorder_count = client.events.aggregate_count(advanced_filter=is_workorder)

Aggregate Event Values Cardinality

EventsAPI.aggregate_cardinality_values(property: cognite.client.data_classes.events.EventProperty | str | list[str], advanced_filter: Optional[Union[Filter, dict[str, Any]]] = None, aggregate_filter: Optional[Union[AggregationFilter, dict[str, Any]]] = None, filter: Optional[Union[EventFilter, dict[str, Any]]] = None) int

Find approximate property count for events.

Parameters
  • property (EventPropertyLike) – The property to count the cardinality of.

  • advanced_filter (Filter | dict[str, Any] | None) – The filter to narrow down the events to count cardinality.

  • aggregate_filter (AggregationFilter | dict[str, Any] | None) – The filter to apply to the resulting buckets.

  • filter (EventFilter | dict[str, Any] | None) – The filter to narrow down the events to count requiring exact match.

Returns

The number of properties matching the specified filter.

Return type

int

Examples:

Count the number of types of events in your CDF project:

>>> from cognite.client import CogniteClient
>>> from cognite.client.data_classes.events import EventProperty
>>> client = CogniteClient()
>>> type_count = client.events.aggregate_cardinality_values(EventProperty.type)

Count the number of types of events linked to asset 123 in your CDF project:

>>> from cognite.client.data_classes import filters
>>> from cognite.client.data_classes.events import EventProperty
>>> is_asset = filters.ContainsAny(EventProperty.asset_ids, 123)
>>> plain_text_author_count = client.events.aggregate_cardinality_values(EventProperty.type, advanced_filter=is_asset)

Aggregate Event Property Cardinality

EventsAPI.aggregate_cardinality_properties(path: cognite.client.data_classes.events.EventProperty | str | list[str], advanced_filter: Optional[Union[Filter, dict[str, Any]]] = None, aggregate_filter: Optional[Union[AggregationFilter, dict[str, Any]]] = None, filter: Optional[Union[EventFilter, dict[str, Any]]] = None) int

Find approximate paths count for events.

Parameters
  • path (EventPropertyLike) – The scope in every document to aggregate properties. The only value allowed now is [“metadata”]. It means to aggregate only metadata properties (aka keys).

  • advanced_filter (Filter | dict[str, Any] | None) – The filter to narrow down the events to count cardinality.

  • aggregate_filter (AggregationFilter | dict[str, Any] | None) – The filter to apply to the resulting buckets.

  • filter (EventFilter | dict[str, Any] | None) – The filter to narrow down the events to count requiring exact match.

Returns

The number of properties matching the specified filters and search.

Return type

int

Examples

Count the number of metadata keys for events in your CDF project:

>>> from cognite.client import CogniteClient
>>> from cognite.client.data_classes.events import EventProperty
>>> client = CogniteClient()
>>> type_count = client.events.aggregate_cardinality_properties(EventProperty.metadata)

Aggregate Event Unique Values

EventsAPI.aggregate_unique_values(filter: Optional[Union[EventFilter, dict[str, Any]]] = None, property: Optional[Union[EventProperty, str, list[str]]] = None, advanced_filter: Optional[Union[Filter, dict[str, Any]]] = None, aggregate_filter: Optional[Union[AggregationFilter, dict[str, Any]]] = None) UniqueResultList

Get unique properties with counts for events.

Parameters
  • filter (EventFilter | dict[str, Any] | None) – The filter to narrow down the events to count requiring exact match.

  • property (EventPropertyLike | None) – The property name(s) to apply the aggregation on.

  • advanced_filter (Filter | dict[str, Any] | None) – The filter to narrow down the events to consider.

  • aggregate_filter (AggregationFilter | dict[str, Any] | None) – The filter to apply to the resulting buckets.

Returns

List of unique values of events matching the specified filters and search.

Return type

UniqueResultList

Examples:

Get the unique types with count of events in your CDF project:

>>> from cognite.client import CogniteClient
>>> from cognite.client.data_classes.events import EventProperty
>>> client = CogniteClient()
>>> result = client.events.aggregate_unique_values(property=EventProperty.type)
>>> print(result.unique)

Get the unique types of events after 2020-01-01 in your CDF project:

>>> from cognite.client.data_classes import filters
>>> from cognite.client.data_classes.events import EventProperty
>>> from cognite.client.utils import timestamp_to_ms
>>> from datetime import datetime
>>> is_after_2020 = filters.Range(EventProperty.start_time, gte=timestamp_to_ms(datetime(2020, 1, 1)))
>>> result = client.events.aggregate_unique_values(EventProperty.type, advanced_filter=is_after_2020)
>>> print(result.unique)

Get the unique types of events after 2020-01-01 in your CDF project, but exclude all types that start with “planned”:

>>> from cognite.client.data_classes.events import EventProperty
>>> from cognite.client.data_classes import aggregations
>>> agg = aggregations
>>> not_planned = agg.Not(agg.Prefix("planned"))
>>> is_after_2020 = filters.Range(EventProperty.start_time, gte=timestamp_to_ms(datetime(2020, 1, 1)))
>>> result = client.events.aggregate_unique_values(EventProperty.type, advanced_filter=is_after_2020, aggregate_filter=not_planned)
>>> print(result.unique)

Aggregate Event Unique Properties

EventsAPI.aggregate_unique_properties(path: cognite.client.data_classes.events.EventProperty | str | list[str], advanced_filter: Optional[Union[Filter, dict[str, Any]]] = None, aggregate_filter: Optional[Union[AggregationFilter, dict[str, Any]]] = None, filter: Optional[Union[EventFilter, dict[str, Any]]] = None) UniqueResultList

Get unique paths with counts for events.

Parameters
  • path (EventPropertyLike) – The scope in every document to aggregate properties. The only value allowed now is [“metadata”]. It means to aggregate only metadata properties (aka keys).

  • advanced_filter (Filter | dict[str, Any] | None) – The filter to narrow down the events to count cardinality.

  • aggregate_filter (AggregationFilter | dict[str, Any] | None) – The filter to apply to the resulting buckets.

  • filter (EventFilter | dict[str, Any] | None) – The filter to narrow down the events to count requiring exact match.

Returns

List of unique values of events matching the specified filters and search.

Return type

UniqueResultList

Examples

Get the unique metadata keys with count of events in your CDF project:

>>> from cognite.client import CogniteClient
>>> from cognite.client.data_classes.events import EventProperty
>>> client = CogniteClient()
>>> result = client.events.aggregate_unique_properties(EventProperty.metadata)
>>> print(result.unique)

Search for events

EventsAPI.search(description: Optional[str] = None, filter: Optional[Union[EventFilter, dict[str, Any]]] = None, limit: int = 25) EventList

Search for events Primarily meant for human-centric use-cases and data exploration, not for programs, since matching and ordering may change over time. Use the list function if stable or exact matches are required.

Parameters
  • description (str | None) – Fuzzy match on description.

  • filter (EventFilter | dict[str, Any] | None) – Filter to apply. Performs exact match on these fields.

  • limit (int) – Maximum number of results to return.

Returns

List of requested events

Return type

EventList

Examples

Search for events:

>>> from cognite.client import CogniteClient
>>> client = CogniteClient()
>>> res = client.events.search(description="some description")

Create events

EventsAPI.create(event: collections.abc.Sequence[cognite.client.data_classes.events.Event] | collections.abc.Sequence[cognite.client.data_classes.events.EventWrite]) EventList
EventsAPI.create(event: cognite.client.data_classes.events.Event | cognite.client.data_classes.events.EventWrite) Event

Create one or more events.

Parameters

event (Event | EventWrite | Sequence[Event] | Sequence[EventWrite]) – Event or list of events to create.

Returns

Created event(s)

Return type

Event | EventList

Examples

Create new events:

>>> from cognite.client import CogniteClient
>>> from cognite.client.data_classes import EventWrite
>>> client = CogniteClient()
>>> events = [EventWrite(start_time=0, end_time=1), EventWrite(start_time=2, end_time=3)]
>>> res = client.events.create(events)

Delete events

EventsAPI.delete(id: Optional[Union[int, Sequence[int]]] = None, external_id: Optional[Union[str, SequenceNotStr[str]]] = None, ignore_unknown_ids: bool = False) None

Delete one or more events

Parameters
  • id (int | Sequence[int] | None) – Id or list of ids

  • external_id (str | SequenceNotStr[str] | None) – External ID or list of external ids

  • ignore_unknown_ids (bool) – Ignore IDs and external IDs that are not found rather than throw an exception.

Examples

Delete events by id or external id:

>>> from cognite.client import CogniteClient
>>> client = CogniteClient()
>>> client.events.delete(id=[1,2,3], external_id="3")

Update events

EventsAPI.update(item: Sequence[cognite.client.data_classes.events.Event | cognite.client.data_classes.events.EventWrite | cognite.client.data_classes.events.EventUpdate], mode: Literal['replace_ignore_null', 'patch', 'replace'] = 'replace_ignore_null') EventList
EventsAPI.update(item: cognite.client.data_classes.events.Event | cognite.client.data_classes.events.EventWrite | cognite.client.data_classes.events.EventUpdate, mode: Literal['replace_ignore_null', 'patch', 'replace'] = 'replace_ignore_null') Event

Update one or more events

Parameters
  • item (Event | EventWrite | EventUpdate | Sequence[Event | EventWrite | EventUpdate]) – Event(s) to update

  • mode (Literal['replace_ignore_null', 'patch', 'replace']) – How to update data when a non-update object is given (Event or -Write). If you use ‘replace_ignore_null’, only the fields you have set will be used to replace existing (default). Using ‘replace’ will additionally clear all the fields that are not specified by you. Last option, ‘patch’, will update only the fields you have set and for container-like fields such as metadata or labels, add the values to the existing. For more details, see Update and Upsert Mode Parameter.

Returns

Updated event(s)

Return type

Event | EventList

Examples

Update an event that you have fetched. This will perform a full update of the event:

>>> from cognite.client import CogniteClient
>>> client = CogniteClient()
>>> event = client.events.retrieve(id=1)
>>> event.description = "New description"
>>> res = client.events.update(event)

Perform a partial update on a event, updating the description and adding a new field to metadata:

>>> from cognite.client.data_classes import EventUpdate
>>> my_update = EventUpdate(id=1).description.set("New description").metadata.add({"key": "value"})
>>> res = client.events.update(my_update)

Upsert events

EventsAPI.upsert(item: Sequence[cognite.client.data_classes.events.Event | cognite.client.data_classes.events.EventWrite], mode: Literal['patch', 'replace'] = 'patch') EventList
EventsAPI.upsert(item: cognite.client.data_classes.events.Event | cognite.client.data_classes.events.EventWrite, mode: Literal['patch', 'replace'] = 'patch') Event
Upsert events, i.e., update if it exists, and create if it does not exist.

Note this is a convenience method that handles the upserting for you by first calling update on all items, and if any of them fail because they do not exist, it will create them instead.

For more details, see Upsert.

Parameters
  • item (Event | EventWrite | Sequence[Event | EventWrite]) – Event or list of events to upsert.

  • mode (Literal['patch', 'replace']) – Whether to patch or replace in the case the events are existing. If you set ‘patch’, the call will only update fields with non-null values (default). Setting ‘replace’ will unset any fields that are not specified.

Returns

The upserted event(s).

Return type

Event | EventList

Examples

Upsert for events:

>>> from cognite.client import CogniteClient
>>> from cognite.client.data_classes import Event
>>> client = CogniteClient()
>>> existing_event = client.events.retrieve(id=1)
>>> existing_event.description = "New description"
>>> new_event = Event(external_id="new_event", description="New event")
>>> res = client.events.upsert([existing_event, new_event], mode="replace")

Filter events

EventsAPI.filter(filter: cognite.client.data_classes.filters.Filter | dict, sort: Optional[Union[EventSort, str, SortableEventProperty, tuple[str, Literal['asc', 'desc']], tuple[str, Literal['asc', 'desc'], Literal['auto', 'first', 'last']], list[cognite.client.data_classes.events.EventSort | str | cognite.client.data_classes.events.SortableEventProperty | tuple[str, Literal['asc', 'desc']] | tuple[str, Literal['asc', 'desc'], Literal['auto', 'first', 'last']]]]] = None, limit: int | None = 25) EventList

Advanced filter events

Advanced filter lets you create complex filtering expressions that combine simple operations, such as equals, prefix, exists, etc., using boolean operators and, or, and not. It applies to basic fields as well as metadata.

Parameters
  • filter (Filter | dict) – Filter to apply.

  • sort (SortSpec | list[SortSpec] | None) – The criteria to sort by. Can be up to two properties to sort by default to ascending order.

  • limit (int | None) – Maximum number of results to return. Defaults to 25. Set to -1, float(“inf”) or None to return all items.

Returns

List of events that match the filter criteria.

Return type

EventList

Examples

Find all events that has external id with prefix “workorder” and the word ‘failure’ in the description, and sort by start time descending:

>>> from cognite.client import CogniteClient
>>> from cognite.client.data_classes import filters
>>> client = CogniteClient()
>>> is_workorder = filters.Prefix("external_id", "workorder")
>>> has_failure = filters.Search("description", "failure")
>>> res = client.events.filter(
...     filter=filters.And(is_workorder, has_failure), sort=("start_time", "desc"))

Note that you can check the API documentation above to see which properties you can filter on with which filters.

To make it easier to avoid spelling mistakes and easier to look up available properties for filtering and sorting, you can also use the EventProperty and SortableEventProperty enums.

>>> from cognite.client.data_classes import filters
>>> from cognite.client.data_classes.events import EventProperty, SortableEventProperty
>>> is_workorder = filters.Prefix(EventProperty.external_id, "workorder")
>>> has_failure = filters.Search(EventProperty.description, "failure")
>>> res = client.events.filter(
...     filter=filters.And(is_workorder, has_failure),
...     sort=(SortableEventProperty.start_time, "desc"))

Events Data classes

class cognite.client.data_classes.events.EndTimeFilter(max: Optional[int] = None, min: Optional[int] = None, is_null: Optional[bool] = None, **_: Any)

Bases: CogniteObject

Either range between two timestamps or isNull filter condition.

Parameters
  • max (int | None) – The number of milliseconds since 00:00:00 Thursday, 1 January 1970, Coordinated Universal Time (UTC), minus leap seconds.

  • min (int | None) – The number of milliseconds since 00:00:00 Thursday, 1 January 1970, Coordinated Universal Time (UTC), minus leap seconds.

  • is_null (bool | None) – Set to true if you want to search for data with field value not set, false to search for cases where some value is present.

  • **_ (Any) – No description.

class cognite.client.data_classes.events.Event(external_id: str | None = None, data_set_id: int | None = None, start_time: int | None = None, end_time: int | None = None, type: str | None = None, subtype: str | None = None, description: str | None = None, metadata: dict[str, str] | None = None, asset_ids: Sequence[int] | None = None, source: str | None = None, id: int | None = None, last_updated_time: int | None = None, created_time: int | None = None, cognite_client: CogniteClient | None = None)

Bases: EventCore

An event represents something that happened at a given interval in time, e.g a failure, a work order etc. This is the reading version of the Event class. It is used when retrieving existing events.

Parameters
  • external_id (str | None) – The external ID provided by the client. Must be unique for the resource type.

  • data_set_id (int | None) – The id of the dataset this event belongs to.

  • start_time (int | None) – The number of milliseconds since 00:00:00 Thursday, 1 January 1970, Coordinated Universal Time (UTC), minus leap seconds.

  • end_time (int | None) – The number of milliseconds since 00:00:00 Thursday, 1 January 1970, Coordinated Universal Time (UTC), minus leap seconds.

  • type (str | None) – Type of the event, e.g. ‘failure’.

  • subtype (str | None) – SubType of the event, e.g. ‘electrical’.

  • description (str | None) – Textual description of the event.

  • metadata (dict[str, str] | None) – Custom, application-specific metadata. String key -> String value. Limits: Maximum length of key is 128 bytes, value 128000 bytes, up to 256 key-value pairs, of total size at most 200000.

  • asset_ids (Sequence[int] | None) – Asset IDs of equipment that this event relates to.

  • source (str | None) – The source of this event.

  • id (int | None) – A server-generated ID for the object.

  • last_updated_time (int | None) – The number of milliseconds since 00:00:00 Thursday, 1 January 1970, Coordinated Universal Time (UTC), minus leap seconds.

  • created_time (int | None) – The number of milliseconds since 00:00:00 Thursday, 1 January 1970, Coordinated Universal Time (UTC), minus leap seconds.

  • cognite_client (CogniteClient | None) – The client to associate with this object.

as_write() EventWrite

Returns this Event in its writing version.

class cognite.client.data_classes.events.EventCore(external_id: Optional[str] = None, data_set_id: Optional[int] = None, start_time: Optional[int] = None, end_time: Optional[int] = None, type: Optional[str] = None, subtype: Optional[str] = None, description: Optional[str] = None, metadata: Optional[dict[str, str]] = None, asset_ids: Optional[Sequence[int]] = None, source: Optional[str] = None)

Bases: WriteableCogniteResource[EventWrite], ABC

An event represents something that happened at a given interval in time, e.g a failure, a work order etc.

Parameters
  • external_id (str | None) – The external ID provided by the client. Must be unique for the resource type.

  • data_set_id (int | None) – The id of the dataset this event belongs to.

  • start_time (int | None) – The number of milliseconds since 00:00:00 Thursday, 1 January 1970, Coordinated Universal Time (UTC), minus leap seconds.

  • end_time (int | None) – The number of milliseconds since 00:00:00 Thursday, 1 January 1970, Coordinated Universal Time (UTC), minus leap seconds.

  • type (str | None) – Type of the event, e.g. ‘failure’.

  • subtype (str | None) – SubType of the event, e.g. ‘electrical’.

  • description (str | None) – Textual description of the event.

  • metadata (dict[str, str] | None) – Custom, application-specific metadata. String key -> String value. Limits: Maximum length of key is 128 bytes, value 128000 bytes, up to 256 key-value pairs, of total size at most 200000.

  • asset_ids (Sequence[int] | None) – Asset IDs of equipment that this event relates to.

  • source (str | None) – The source of this event.

class cognite.client.data_classes.events.EventFilter(start_time: Optional[Union[dict[str, Any], TimestampRange]] = None, end_time: Optional[Union[dict[str, Any], EndTimeFilter]] = None, active_at_time: Optional[Union[dict[str, Any], TimestampRange]] = None, metadata: Optional[dict[str, str]] = None, asset_ids: Optional[Sequence[int]] = None, asset_external_ids: Optional[SequenceNotStr[str]] = None, asset_subtree_ids: Optional[Sequence[dict[str, Any]]] = None, data_set_ids: Optional[Sequence[dict[str, Any]]] = None, source: Optional[str] = None, type: Optional[str] = None, subtype: Optional[str] = None, created_time: Optional[Union[dict[str, Any], TimestampRange]] = None, last_updated_time: Optional[Union[dict[str, Any], TimestampRange]] = None, external_id_prefix: Optional[str] = None)

Bases: CogniteFilter

Filter on events filter with exact match

Parameters
  • start_time (dict[str, Any] | TimestampRange | None) – Range between two timestamps.

  • end_time (dict[str, Any] | EndTimeFilter | None) – Either range between two timestamps or isNull filter condition.

  • active_at_time (dict[str, Any] | TimestampRange | None) – Event is considered active from its startTime to endTime inclusive. If startTime is null, event is never active. If endTime is null, event is active from startTime onwards. activeAtTime filter will match all events that are active at some point from min to max, from min, or to max, depending on which of min and max parameters are specified.

  • metadata (dict[str, str] | None) – Custom, application specific metadata. String key -> String value. Limits: Maximum length of key is 128 bytes, value 128000 bytes, up to 256 key-value pairs, of total size at most 200000.

  • asset_ids (Sequence[int] | None) – Asset IDs of equipment that this event relates to.

  • asset_external_ids (SequenceNotStr[str] | None) – Asset External IDs of equipment that this event relates to.

  • asset_subtree_ids (Sequence[dict[str, Any]] | None) – Only include events that have a related asset in a subtree rooted at any of these assetIds. If the total size of the given subtrees exceeds 100,000 assets, an error will be returned.

  • data_set_ids (Sequence[dict[str, Any]] | None) – Only include events that belong to these datasets.

  • source (str | None) – The source of this event.

  • type (str | None) – Type of the event, e.g ‘failure’.

  • subtype (str | None) – SubType of the event, e.g ‘electrical’.

  • created_time (dict[str, Any] | TimestampRange | None) – Range between two timestamps.

  • last_updated_time (dict[str, Any] | TimestampRange | None) – Range between two timestamps.

  • external_id_prefix (str | None) – Filter by this (case-sensitive) prefix for the external ID.

dump(camel_case: bool = True) dict[str, Any]

Dump the instance into a json serializable Python data type.

Parameters

camel_case (bool) – Use camelCase for attribute names. Defaults to True.

Returns

A dictionary representation of the instance.

Return type

dict[str, Any]

class cognite.client.data_classes.events.EventList(resources: Iterable[Any], cognite_client: CogniteClient | None = None)

Bases: WriteableCogniteResourceList[EventWrite, Event], IdTransformerMixin

class cognite.client.data_classes.events.EventProperty(value)

Bases: EnumProperty

An enumeration.

class cognite.client.data_classes.events.EventUpdate(id: Optional[int] = None, external_id: Optional[str] = None)

Bases: CogniteUpdate

Changes will be applied to event.

Parameters
  • id (int) – A server-generated ID for the object.

  • external_id (str) – The external ID provided by the client. Must be unique for the resource type.

class cognite.client.data_classes.events.EventWrite(external_id: Optional[str] = None, data_set_id: Optional[int] = None, start_time: Optional[int] = None, end_time: Optional[int] = None, type: Optional[str] = None, subtype: Optional[str] = None, description: Optional[str] = None, metadata: Optional[dict[str, str]] = None, asset_ids: Optional[Sequence[int]] = None, source: Optional[str] = None)

Bases: EventCore

An event represents something that happened at a given interval in time, e.g a failure, a work order etc. This is the writing version of the Event class. It is used when creating new events.

Parameters
  • external_id (str | None) – The external ID provided by the client. Must be unique for the resource type.

  • data_set_id (int | None) – The id of the dataset this event belongs to.

  • start_time (int | None) – The number of milliseconds since 00:00:00 Thursday, 1 January 1970, Coordinated Universal Time (UTC), minus leap seconds.

  • end_time (int | None) – The number of milliseconds since 00:00:00 Thursday, 1 January 1970, Coordinated Universal Time (UTC), minus leap seconds.

  • type (str | None) – Type of the event, e.g. ‘failure’.

  • subtype (str | None) – SubType of the event, e.g. ‘electrical’.

  • description (str | None) – Textual description of the event.

  • metadata (dict[str, str] | None) – Custom, application-specific metadata. String key -> String value. Limits: Maximum length of key is 128 bytes, value 128000 bytes, up to 256 key-value pairs, of total size at most 200000.

  • asset_ids (Sequence[int] | None) – Asset IDs of equipment that this event relates to.

  • source (str | None) – The source of this event.

as_write() EventWrite

Returns self.

class cognite.client.data_classes.events.EventWriteList(resources: Iterable[Any], cognite_client: CogniteClient | None = None)

Bases: CogniteResourceList[EventWrite], ExternalIDTransformerMixin

class cognite.client.data_classes.events.SortableEventProperty(value)

Bases: EnumProperty

An enumeration.