Data Modeling

Data Models

Retrieve data models by id(s)

DataModelsAPI.retrieve(ids: DataModelIdentifier | Sequence[DataModelIdentifier], inline_views: Literal[True]) → DataModelList[View]

DataModelsAPI.retrieve(ids: DataModelIdentifier | Sequence[DataModelIdentifier], inline_views: Literal[False] = False) → DataModelList[ViewId]

Retrieve data_model(s) by id(s).

Parameters

ids (DataModelIdentifier | Sequence[DataModelIdentifier]) – Data Model identifier(s).
inline_views (bool) – Whether to expand the referenced views inline in the returned result.

Returns

Requested data_model or None if it does not exist.

Return type

DataModelList[ViewId] | DataModelList[View]

Examples

>>> from cognite.client import CogniteClient
>>> c = CogniteClient()
>>> res = c.data_modeling.data_models.retrieve(("mySpace", "myDataModel", "v1"))

List data models

DataModelsAPI.list(inline_views: Literal[True], limit: int | None = DATA_MODELING_DEFAULT_LIMIT_READ, space: str | None = None, all_versions: bool = False, include_global: bool = False) → DataModelList[View]

DataModelsAPI.list(inline_views: Literal[False] = False, limit: int | None = DATA_MODELING_DEFAULT_LIMIT_READ, space: str | None = None, all_versions: bool = False, include_global: bool = False) → DataModelList[ViewId]

List data models

Parameters

inline_views (bool) – Whether to expand the referenced views inline in the returned result.
limit (int | None) – Maximum number of data model to return. Defaults to 10. Set to -1, float(“inf”) or None to return all items.
space (str | None) – The space to query.
all_versions (bool) – Whether to return all versions. If false, only the newest version is returned, which is determined based on the ‘createdTime’ field.
include_global (bool) – Whether to include global data models.

Returns

List of requested data models

Return type

DataModelList[View] | DataModelList[ViewId]

Examples

List 5 data model:

>>> from cognite.client import CogniteClient
>>> c = CogniteClient()
>>> data_model_list = c.data_modeling.data_models.list(limit=5)

Iterate over data model:

>>> from cognite.client import CogniteClient
>>> c = CogniteClient()
>>> for data_model in c.data_modeling.data_models:
...     data_model # do something with the data_model

Iterate over chunks of data model to reduce memory load:

>>> from cognite.client import CogniteClient
>>> c = CogniteClient()
>>> for data_model_list in c.data_modeling.data_models(chunk_size=10):
...     data_model_list # do something with the data model

Apply data models

DataModelsAPI.apply(data_model: Sequence[DataModelApply]) → DataModelList

DataModelsAPI.apply(data_model: DataModelApply) → DataModel

Create or update one or more data model.

Parameters: data_model (DataModelApply | Sequence[DataModelApply]) – Data model(s) to create or update (upsert).
Returns: Created data model(s)
Return type: DataModel | DataModelList

Examples

Create new data model:

>>> from cognite.client import CogniteClient
>>> from cognite.client.data_classes.data_modeling import DataModelApply
>>> c = CogniteClient()
>>> data_models = [DataModelApply(space="mySpace",external_id="myDataModel",version="v1"),
... DataModelApply(space="mySpace",external_id="myOtherDataModel",version="v1")]
>>> res = c.data_modeling.data_models.apply(data_models)

Delete data models

DataModelsAPI.delete(ids: DataModelIdentifier | Sequence[DataModelIdentifier]) → list[DataModelId]

Delete one or more data model

Parameters: ids (DataModelIdentifier | Sequence[DataModelIdentifier]) – Data Model identifier(s).
Returns: The data_model(s) which has been deleted. None if nothing was deleted.
Return type: list[DataModelId]

Examples

Delete data model by id:

>>> from cognite.client import CogniteClient
>>> c = CogniteClient()
>>> c.data_modeling.data_models.delete(("mySpace", "myDataModel", "v1"))

Data model data classes

class cognite.client.data_classes.data_modeling.data_models.DataModel(space: str, external_id: str, version: str, is_global: bool, last_updated_time: int, created_time: int, description: str | None = None, name: str | None = None, views: list[T_View] | None = None, **_: Any)

Bases: DataModelCore, Generic[T_View]

A group of views. This is the read version of a Data Model

Parameters

space (str) – The workspace for the data model, a unique identifier for the space.
external_id (str) – Combined with the space is the unique identifier of the data model.
version (str) – DMS version.
is_global (bool) – Whether this is a global data model.
last_updated_time (int) – The number of milliseconds since 00:00:00 Thursday, 1 January 1970, Coordinated Universal Time (UTC), minus leap seconds.
created_time (int) – The number of milliseconds since 00:00:00 Thursday, 1 January 1970, Coordinated Universal Time (UTC), minus leap seconds.
description (str | None) – Textual description of the data model
name (str | None) – Human readable name for the data model.
views (list[T_View] | None) – List of views included in this data model.
**_ (Any) – No description.

dump(camel_case: bool = False) → dict[str, Any]

Dump the instance into a json serializable Python data type.

Parameters: camel_case (bool) – Use camelCase for attribute names. Defaults to False.
Returns: A dictionary representation of the instance.
Return type: dict[str, Any]

class cognite.client.data_classes.data_modeling.data_models.DataModelApply(space: str, external_id: str, version: str, description: str | None = None, name: str | None = None, views: list[ViewId | ViewApply] | None = None)

Bases: DataModelCore

A group of views. This is the write version of a Data Model.

Parameters

space (str) – The workspace for the data model, a unique identifier for the space.
external_id (str) – Combined with the space is the unique identifier of the data model.
version (str) – DMS version.
description (str | None) – Textual description of the data model
name (str | None) – Human readable name for the data model.
views (list[ViewId | ViewApply] | None) – List of views included in this data model.

dump(camel_case: bool = False) → dict[str, Any]

Dump the instance into a json serializable Python data type.

Parameters: camel_case (bool) – Use camelCase for attribute names. Defaults to False.
Returns: A dictionary representation of the instance.
Return type: dict[str, Any]

class cognite.client.data_classes.data_modeling.data_models.DataModelApplyList(resources: Collection[Any], cognite_client: CogniteClient | None = None)

Bases: CogniteResourceList[DataModelApply]

as_ids() → list[DataModelId]

Convert the list of data models to a list of data model ids.

Returns: The list of data model ids.
Return type: list[DataModelId]

class cognite.client.data_classes.data_modeling.data_models.DataModelCore(space: str, external_id: str, version: str, description: str | None = None, name: str | None = None, **_: Any)

Bases: DataModelingResource

A group of views.

Parameters

space (str) – The workspace for the data model, a unique identifier for the space.
external_id (str) – Combined with the space is the unique identifier of the data model.
version (str) – DMS version.
description (str | None) – Textual description of the data model
name (str | None) – Human readable name for the data model.
**_ (Any) – No description.

class cognite.client.data_classes.data_modeling.data_models.DataModelFilter(space: str | None = None, inline_views: bool = False, all_versions: bool = False, include_global: bool = False)

Bases: CogniteFilter

Represent the filer arguments for the list endpoint.

Parameters

space (str | None) – The space to query
inline_views (bool) – Whether to expand the referenced views inline in the returned result.
all_versions (bool) – Whether to return all versions. If false, only the newest version is returned, which is determined based on the ‘createdTime’ field.
include_global (bool) – Whether to include global views.

class cognite.client.data_classes.data_modeling.data_models.DataModelList(resources: Collection[Any], cognite_client: CogniteClient | None = None)

Bases: CogniteResourceList[DataModel[T_View]]

as_apply() → DataModelApplyList

Convert the list of data models to a list of data model applies.

Returns: The list of data model applies.
Return type: DataModelApplyList

as_ids() → list[DataModelId]

Convert the list of data models to a list of data model ids.

Returns: The list of data model ids.
Return type: list[DataModelId]

latest_version(key: Literal['created_time', 'last_updated_time'] = 'created_time') → DataModel[T_View]

Get the data model in the list with the latest version. The latest version is determined based on the created_time or last_updated_time field.

Parameters: key (Literal["created_time", "last_updated_time"]) – The field to use for determining the latest version.
Returns: The data model with the latest version.
Return type: DataModel[T_View]

Spaces

Retrieve a space by id

SpacesAPI.retrieve(space: str) → Space | None

SpacesAPI.retrieve(space: Sequence[str]) → SpaceList

Retrieve space by id.

Parameters: space (str | Sequence[str]) – Space ID
Returns: Requested space or None if it does not exist.
Return type: Space | SpaceList | None

Examples

>>> from cognite.client import CogniteClient
>>> c = CogniteClient()
>>> res = c.data_modeling.spaces.retrieve(space='mySpace')

Get multiple spaces by id:

>>> from cognite.client import CogniteClient
>>> c = CogniteClient()
>>> res = c.data_modeling.spaces.retrieve(spaces=["MySpace", "MyAwesomeSpace", "MyOtherSpace"])

List spaces

SpacesAPI.list(limit: int | None = 25, include_global: bool = False) → SpaceList

List spaces

Parameters

limit (int | None) – Maximum number of spaces to return. Defaults to 10. Set to -1, float(“inf”) or None to return all items.
include_global (bool) – Whether to include global spaces. Defaults to False.

Returns

List of requested spaces

Return type

SpaceList

Examples

List spaces and filter on max start time:

>>> from cognite.client import CogniteClient
>>> c = CogniteClient()
>>> space_list = c.data_modeling.spaces.list(limit=5)

Iterate over spaces:

>>> from cognite.client import CogniteClient
>>> c = CogniteClient()
>>> for space in c.data_modeling.spaces:
...     space # do something with the space

Iterate over chunks of spaces to reduce memory load:

>>> from cognite.client import CogniteClient
>>> c = CogniteClient()
>>> for space_list in c.data_modeling.spaces(chunk_size=2500):
...     space_list # do something with the spaces

Apply spaces

SpacesAPI.apply(space: Sequence[SpaceApply]) → SpaceList

SpacesAPI.apply(space: SpaceApply) → Space

Create or patch one or more spaces.

Parameters: space (SpaceApply | Sequence[SpaceApply]) – Space | Sequence[Space]): Space or spaces of spacesda to create or update.
Returns: Created space(s)
Return type: Space | SpaceList

Examples

Create new spaces:

>>> from cognite.client import CogniteClient
>>> from cognite.client.data_classes.data_modeling import SpaceApply
>>> c = CogniteClient()
>>> spaces = [SpaceApply(space="mySpace", description="My first space", name="My Space"),
... SpaceApply(space="myOtherSpace", description="My second space", name="My Other Space")]
>>> res = c.data_modeling.spaces.apply(spaces)

Delete spaces

SpacesAPI.delete(space: str | Sequence[str]) → list[str]

Delete one or more spaces

Parameters: space (str | Sequence[str]) – ID or ID list ids of spaces.
Returns: The space(s) which has been deleted.
Return type: list[str]

Examples

Delete spaces by id:

>>> from cognite.client import CogniteClient
>>> c = CogniteClient()
>>> c.data_modeling.spaces.delete(space=["mySpace", "myOtherSpace"])

Data classes

class cognite.client.data_classes.data_modeling.spaces.Space(space: str, is_global: bool, last_updated_time: int, created_time: int, description: str | None = None, name: str | None = None, **_: Any)

Bases: SpaceCore

A workspace for data models and instances. This is the read version.

Parameters

space (str) – A unique identifier for the space.
is_global (bool) – Whether the space is global or not.
last_updated_time (int) – The number of milliseconds since 00:00:00 Thursday, 1 January 1970, Coordinated Universal Time (UTC), minus leap seconds.
created_time (int) – The number of milliseconds since 00:00:00 Thursday, 1 January 1970, Coordinated Universal Time (UTC), minus leap seconds.
description (str | None) – Textual description of the space
name (str | None) – Human readable name for the space.
**_ (Any) – No description.

class cognite.client.data_classes.data_modeling.spaces.SpaceApply(space: str, description: str | None = None, name: str | None = None, **_: Any)

Bases: SpaceCore

A workspace for data models and instances. This is the write version

Parameters

space (str) – A unique identifier for the space.
description (str | None) – Textual description of the space
name (str | None) – Human readable name for the space.
**_ (Any) – No description.

class cognite.client.data_classes.data_modeling.spaces.SpaceApplyList(resources: Collection[Any], cognite_client: CogniteClient | None = None)

Bases: CogniteResourceList[SpaceApply]

as_ids() → list[str]

Converts all the spaces to a space id list.

Returns: A list of space ids.
Return type: list[str]

class cognite.client.data_classes.data_modeling.spaces.SpaceCore(space: str, description: str | None = None, name: str | None = None)

Bases: DataModelingResource

A workspace for data models and instances.

Parameters

space (str) – A unique identifier for the space.
description (str | None) – Textual description of the space
name (str | None) – Human readable name for the space.

class cognite.client.data_classes.data_modeling.spaces.SpaceList(resources: Collection[Any], cognite_client: CogniteClient | None = None)

Bases: CogniteResourceList[Space]

as_apply() → SpaceApplyList

Converts all the spaces to a space apply list.

Returns: A list of space applies.
Return type: SpaceApplyList

as_ids() → list[str]

Converts all the spaces to a space id list..

Returns: A list of space ids.
Return type: list[str]

Views

Retrieve views by id(s)

ViewsAPI.retrieve(ids: ViewIdentifier | Sequence[ViewIdentifier], include_inherited_properties: bool = True, all_versions: bool = True) → ViewList

Retrieve a single view by id.

Parameters

ids (ViewIdentifier | Sequence[ViewIdentifier]) – View identifier(s)
include_inherited_properties (bool) – Whether to include properties inherited from views this view implements.
all_versions (bool) – Whether to return all versions. If false, only the newest version is returned,

Returns

Requested view or None if it does not exist.

Return type

ViewList

Examples

>>> from cognite.client import CogniteClient
>>> c = CogniteClient()
>>> res = c.data_modeling.views.retrieve(('mySpace', 'myView', 'v1'))

List views

ViewsAPI.list(limit: int | None = 10, space: str | None = None, include_inherited_properties: bool = True, all_versions: bool = False, include_global: bool = False) → ViewList

List views

Parameters

limit (int | None) – Maximum number of views to return. Defaults to 10. Set to -1, float(“inf”) or None to return all items.
space (str | None) – (str | None): The space to query.
include_inherited_properties (bool) – Whether to include properties inherited from views this view implements.
all_versions (bool) – Whether to return all versions. If false, only the newest version is returned, which is determined based on the ‘createdTime’ field.
include_global (bool) – Whether to include global views.

Returns

List of requested views

Return type

ViewList

Examples

List 5 views:

>>> from cognite.client import CogniteClient
>>> c = CogniteClient()
>>> view_list = c.data_modeling.views.list(limit=5)

Iterate over views:

>>> from cognite.client import CogniteClient
>>> c = CogniteClient()
>>> for view in c.data_modeling.views:
...     view # do something with the view

Iterate over chunks of views to reduce memory load:

>>> from cognite.client import CogniteClient
>>> c = CogniteClient()
>>> for view_list in c.data_modeling.views(chunk_size=10):
...     view_list # do something with the views

Apply view

ViewsAPI.apply(view: Sequence[ViewApply]) → ViewList

ViewsAPI.apply(view: ViewApply) → View

Create or update (upsert) one or more views.

Parameters: view (ViewApply | Sequence[ViewApply]) – View(s) to create or update.
Returns: Created view(s)
Return type: View | ViewList

Examples

Create new views:

>>> from cognite.client import CogniteClient
>>> from cognite.client.data_classes.data_modeling import ViewApply
>>> c = CogniteClient()
>>> views = [ViewApply(space="mySpace",external_id="myView",version="v1"),
... ViewApply(space="mySpace",external_id="myOtherView",version="v1")]
>>> res = c.data_modeling.views.apply(views)

Delete views

ViewsAPI.delete(ids: ViewIdentifier | Sequence[ViewIdentifier]) → list[ViewId]

Delete one or more views

Parameters: ids (ViewIdentifier | Sequence[ViewIdentifier]) – View identifier(s)
Returns: The identifier for the view(s) which has been deleted. Empty list if nothing was deleted.
Return type: list[ViewId]

Examples

Delete views by id:

>>> from cognite.client import CogniteClient
>>> c = CogniteClient()
>>> c.data_modeling.views.delete(('mySpace', 'myView', 'v1'))

View data classes

class cognite.client.data_classes.data_modeling.views.ConnectionDefinition: Bases: ViewProperty

class cognite.client.data_classes.data_modeling.views.ConnectionDefinitionApply: Bases: ViewPropertyApply

class cognite.client.data_classes.data_modeling.views.MappedProperty(container: 'ContainerId', container_property_identifier: 'str', type: 'PropertyType', nullable: 'bool', auto_increment: 'bool', source: 'ViewId | None' = None, default_value: 'str | int | dict | None' = None, name: 'str | None' = None, description: 'str | None' = None): Bases: ViewProperty

class cognite.client.data_classes.data_modeling.views.MappedPropertyApply(container: 'ContainerId', container_property_identifier: 'str', name: 'str | None' = None, description: 'str | None' = None, source: 'ViewId | None' = None): Bases: ViewPropertyApply

class cognite.client.data_classes.data_modeling.views.SingleHopConnectionDefinition(type: 'DirectRelationReference', source: 'ViewId', name: 'str | None' = None, description: 'str | None' = None, edge_source: 'ViewId | None' = None, direction: "Literal[('outwards', 'inwards')]" = 'outwards', connection_type: "Literal['multiEdgeConnection']" = 'multiEdgeConnection'): Bases: ConnectionDefinition

class cognite.client.data_classes.data_modeling.views.SingleHopConnectionDefinitionApply(type: 'DirectRelationReference', source: 'ViewId', name: 'str | None' = None, description: 'str | None' = None, edge_source: 'ViewId | None' = None, direction: "Literal[('outwards', 'inwards')]" = 'outwards', connection_type: "Literal['multiEdgeConnection']" = 'multiEdgeConnection'): Bases: ConnectionDefinitionApply

class cognite.client.data_classes.data_modeling.views.View(space: str, external_id: str, version: str, properties: dict[str, MappedProperty | ConnectionDefinition], last_updated_time: int, created_time: int, description: str | None = None, name: str | None = None, filter: Filter | None = None, implements: list[ViewId] | None = None, writable: bool = False, used_for: Literal['node', 'edge', 'all'] = 'node', is_global: bool = False, **_: Any)

Bases: ViewCore

A group of properties. Read only version.

Parameters

space (str) – The workspace for the view, a unique identifier for the space.
external_id (str) – Combined with the space is the unique identifier of the view.
version (str) – DMS version.
properties (dict[str, MappedProperty | ConnectionDefinition]) – View with included properties and expected edges, indexed by a unique space-local identifier.
last_updated_time (int) – The number of milliseconds since 00:00:00 Thursday, 1 January 1970, Coordinated Universal Time (UTC), minus leap seconds.
created_time (int) – The number of milliseconds since 00:00:00 Thursday, 1 January 1970, Coordinated Universal Time (UTC), minus leap seconds.
description (str | None) – Textual description of the view
name (str | None) – Human readable name for the view.
filter (Filter | None) – A filter Domain Specific Language (DSL) used to create advanced filter queries.
implements (list[ViewId] | None) – References to the views from where this view will inherit properties and edges.
writable (bool) – Whether the view supports write operations.
used_for (Literal["node", "edge", "all"]) – Does this view apply to nodes, edges or both.
is_global (bool) – Whether this is a global container, i.e., one of the out-of-the-box models.
**_ (Any) – No description.

as_apply() → ViewApply

Convert to a view applies.

Returns: The view apply.
Return type: ViewApply

dump(camel_case: bool = False) → dict[str, Any]

Dump the instance into a json serializable Python data type.

Parameters: camel_case (bool) – Use camelCase for attribute names. Defaults to False.
Returns: A dictionary representation of the instance.
Return type: dict[str, Any]

class cognite.client.data_classes.data_modeling.views.ViewApply(space: str, external_id: str, version: str, description: str | None = None, name: str | None = None, filter: Filter | None = None, implements: list[ViewId] | None = None, properties: dict[str, MappedPropertyApply | ConnectionDefinitionApply] | None = None)

Bases: ViewCore

A group of properties. Write only version.

Parameters

space (str) – The workspace for the view, a unique identifier for the space.
external_id (str) – Combined with the space is the unique identifier of the view.
version (str) – DMS version.
description (str | None) – Textual description of the view
name (str | None) – Human readable name for the view.
filter (Filter | None) – A filter Domain Specific Language (DSL) used to create advanced filter queries.
implements (list[ViewId] | None) – References to the views from where this view will inherit properties and edges.
properties (dict[str, MappedPropertyApply | ConnectionDefinitionApply] | None) – No description.

dump(camel_case: bool = False) → dict[str, Any]

Dump the instance into a json serializable Python data type.

Parameters: camel_case (bool) – Use camelCase for attribute names. Defaults to False.
Returns: A dictionary representation of the instance.
Return type: dict[str, Any]

class cognite.client.data_classes.data_modeling.views.ViewApplyList(resources: Collection[Any], cognite_client: CogniteClient | None = None)

Bases: CogniteResourceList[ViewApply]

as_ids() → list[ViewId]

Returns the list of ViewIds

Returns: The list of ViewIds
Return type: list[ViewId]

class cognite.client.data_classes.data_modeling.views.ViewFilter(space: str | None = None, include_inherited_properties: bool = True, all_versions: bool = False, include_global: bool = False)

Bases: CogniteFilter

Represent the filer arguments for the list endpoint.

Parameters

space (str | None) – The space to query
include_inherited_properties (bool) – Whether to include properties inherited from views this view implements.
all_versions (bool) – Whether to return all versions. If false, only the newest version is returned, which is determined based on the ‘createdTime’ field.
include_global (bool) – Whether to include global views.

class cognite.client.data_classes.data_modeling.views.ViewList(resources: Collection[Any], cognite_client: CogniteClient | None = None)

Bases: CogniteResourceList[View]

as_apply() → ViewApplyList

Convert to a view an apply list.

Returns: The view apply list.
Return type: ViewApplyList

as_ids() → list[ViewId]

Returns the list of ViewIds

Returns: The list of ViewIds
Return type: list[ViewId]

class cognite.client.data_classes.data_modeling.views.ViewProperty: Bases: ABC

class cognite.client.data_classes.data_modeling.views.ViewPropertyApply: Bases: ABC

Containers

Retrieve containers by id(s)

ContainersAPI.retrieve(ids: Union[ContainerId, Tuple[str, str]]) → Container | None

ContainersAPI.retrieve(ids: Sequence[Union[ContainerId, Tuple[str, str]]]) → ContainerList

Retrieve one or more container by id(s).

Parameters: ids (ContainerIdentifier | Sequence[ContainerIdentifier]) – Identifier for container(s).
Returns: Requested container or None if it does not exist.
Return type: Container | ContainerList | None

Examples

>>> from cognite.client import CogniteClient
>>> c = CogniteClient()
>>> res = c.data_modeling.containers.retrieve(('mySpace', 'myContainer'))

Fetch using the ContainerId:

>>> from cognite.client import CogniteClient
>>> from cognite.client.data_classes.data_modeling import ContainerId
>>> c = CogniteClient()
>>> res = c.data_modeling.containers.retrieve(ContainerId(space='mySpace', external_id='myContainer'))

List containers

ContainersAPI.list(space: str | None = None, limit: int | None = 10, include_global: bool = False) → ContainerList

List containers

Parameters

space (str | None) – The space to query
limit (int | None) – Maximum number of containers to return. Defaults to 10. Set to -1, float(“inf”) or None to return all items.
include_global (bool) – Whether the global containers should be returned.

Returns

List of requested containers

Return type

ContainerList

Examples

List containers and limit to 5:

>>> from cognite.client import CogniteClient
>>> c = CogniteClient()
>>> container_list = c.data_modeling.containers.list(limit=5)

Iterate over containers:

>>> from cognite.client import CogniteClient
>>> c = CogniteClient()
>>> for container in c.data_modeling.containers:
...     container # do something with the container

Iterate over chunks of containers to reduce memory load:

>>> from cognite.client import CogniteClient
>>> c = CogniteClient()
>>> for container_list in c.data_modeling.containers(chunk_size=10):
...     container_list # do something with the containers

Apply containers

ContainersAPI.apply(container: Sequence[ContainerApply]) → ContainerList

ContainersAPI.apply(container: ContainerApply) → Container

Add or update (upsert) containers.

Parameters: container (ContainerApply | Sequence[ContainerApply]) – Container(s) to create or update.
Returns: Created container(s)
Return type: Container | ContainerList

Examples

Create new containers:

>>> from cognite.client import CogniteClient
>>> from cognite.client.data_classes.data_modeling import ContainerApply, ContainerProperty, Text
>>> c = CogniteClient()
>>> container = [ContainerApply(space="mySpace", external_id="myContainer",
...     properties={"name": ContainerProperty(type=Text(), name="name")})]
>>> res = c.data_modeling.containers.apply(container)

Delete containers

ContainersAPI.delete(id: ContainerIdentifier | Sequence[ContainerIdentifier]) → list[ContainerId]

Delete one or more containers

Parameters: id (ContainerIdentifier | Sequence[ContainerIdentifier]) – The container identifier(s).
Returns: The container(s) which has been deleted. Empty list if nothing was deleted.
Return type: list[ContainerId]

Examples

Delete containers by id:

>>> from cognite.client import CogniteClient
>>> c = CogniteClient()
>>> c.data_modeling.containers.delete(("mySpace", "myContainer"))

Containers data classes

class cognite.client.data_classes.data_modeling.containers.BTreeIndex(properties: 'list[str]', cursorable: 'bool' = False): Bases: Index

class cognite.client.data_classes.data_modeling.containers.Constraint: Bases: ABC

class cognite.client.data_classes.data_modeling.containers.Container(space: str, external_id: str, properties: dict[str, ContainerProperty], is_global: bool, last_updated_time: int, created_time: int, description: str | None = None, name: str | None = None, used_for: Literal['node', 'edge', 'all'] = 'node', constraints: dict[str, Constraint] | None = None, indexes: dict[str, Index] | None = None, **_: Any)

Bases: ContainerCore

Represent the physical storage of data. This is the read format of the container

Parameters

space (str) – The workspace for the container, a unique identifier for the space.
external_id (str) – Combined with the space is the unique identifier of the view.
properties (dict[str, ContainerProperty]) – We index the property by a local unique identifier.
is_global (bool) – Whether this is a global container, i.e., one of the out-of-the-box models.
last_updated_time (int) – The number of milliseconds since 00:00:00 Thursday, 1 January 1970, Coordinated Universal Time (UTC), minus leap seconds.
created_time (int) – The number of milliseconds since 00:00:00 Thursday, 1 January 1970, Coordinated Universal Time (UTC), minus leap seconds.
description (str | None) – Textual description of the view
name (str | None) – Human readable name for the view.
used_for (Literal["node", "edge", "all"]) – Should this operation apply to nodes, edges or both.
constraints (dict[str, Constraint] | None) – Set of constraints to apply to the container
indexes (dict[str, Index] | None) – Set of indexes to apply to the container.
**_ (Any) – No description.

class cognite.client.data_classes.data_modeling.containers.ContainerApply(space: str, external_id: str, properties: dict[str, ContainerProperty], description: str | None = None, name: str | None = None, used_for: Literal['node', 'edge', 'all'] | None = None, constraints: dict[str, Constraint] | None = None, indexes: dict[str, Index] | None = None)

Bases: ContainerCore

Represent the physical storage of data. This is the write format of the container

Parameters

space (str) – The workspace for the container, a unique identifier for the space.
external_id (str) – Combined with the space is the unique identifier of the view.
properties (dict[str, ContainerProperty]) – We index the property by a local unique identifier.
description (str | None) – Textual description of the view
name (str | None) – Human readable name for the view.
used_for (Literal["node", "edge", "all"] | None) – Should this operation apply to nodes, edges or both.
constraints (dict[str, Constraint] | None) – Set of constraints to apply to the container
indexes (dict[str, Index] | None) – Set of indexes to apply to the container.

class cognite.client.data_classes.data_modeling.containers.ContainerApplyList(resources: Collection[Any], cognite_client: CogniteClient | None = None)

Bases: CogniteResourceList[ContainerApply]

as_ids() → list[ContainerId]

Convert to a container id list.

Returns: The container id list.
Return type: list[ContainerId]

class cognite.client.data_classes.data_modeling.containers.ContainerCore(space: str, external_id: str, properties: dict[str, ContainerProperty], description: str | None = None, name: str | None = None, used_for: Literal['node', 'edge', 'all'] | None = None, constraints: dict[str, Constraint] | None = None, indexes: dict[str, Index] | None = None, **_: Any)

Bases: DataModelingResource

Represent the physical storage of data. This is the base class for the read and write version.

Parameters

space (str) – The workspace for the container, a unique identifier for the space.
external_id (str) – Combined with the space is the unique identifier of the view.
properties (dict[str, ContainerProperty]) – We index the property by a local unique identifier.
description (str | None) – Textual description of the view
name (str | None) – Human readable name for the view.
used_for (Literal["node", "edge", "all"] | None) – Should this operation apply to nodes, edges or both.
constraints (dict[str, Constraint] | None) – Set of constraints to apply to the container
indexes (dict[str, Index] | None) – Set of indexes to apply to the container.
**_ (Any) – No description.

dump(camel_case: bool = False) → dict[str, Any]

Dump the instance into a json serializable Python data type.

Parameters: camel_case (bool) – Use camelCase for attribute names. Defaults to False.
Returns: A dictionary representation of the instance.
Return type: dict[str, Any]

class cognite.client.data_classes.data_modeling.containers.ContainerFilter(space: str | None = None, include_global: bool = False)

Bases: CogniteFilter

Represent the filter arguments for the list endpoint.

Parameters

space (str | None) – The space to query
include_global (bool) – Whether the global containers should be included.

class cognite.client.data_classes.data_modeling.containers.ContainerList(resources: Collection[Any], cognite_client: CogniteClient | None = None)

Bases: CogniteResourceList[Container]

as_apply() → ContainerApplyList

Convert to a ContainerApply list.

Returns: The container apply list.
Return type: ContainerApplyList

as_ids() → list[ContainerId]

Convert to a container id list.

Returns: The container id list.
Return type: list[ContainerId]

class cognite.client.data_classes.data_modeling.containers.ContainerProperty(type: 'PropertyType', nullable: 'bool' = True, auto_increment: 'bool' = False, name: 'str | None' = None, default_value: 'str | int | dict | None' = None, description: 'str | None' = None): Bases: object

class cognite.client.data_classes.data_modeling.containers.Index: Bases: ABC

class cognite.client.data_classes.data_modeling.containers.InvertedIndex(properties: 'list[str]'): Bases: Index

class cognite.client.data_classes.data_modeling.containers.RequiresConstraint(require: 'ContainerId'): Bases: Constraint

cognite.client.data_classes.data_modeling.containers.RequiresConstraintDefinition: alias of RequiresConstraint

class cognite.client.data_classes.data_modeling.containers.UniquenessConstraint(properties: 'list[str]'): Bases: Constraint

cognite.client.data_classes.data_modeling.containers.UniquenessConstraintDefinition: alias of UniquenessConstraint

Instances

Retrieve instances by id(s)

Retrieve one or more instance by id(s).

Parameters

nodes (NodeId | Sequence[NodeId] | tuple[str, str] | Sequence[tuple[str, str]] | None) – Node ids
edges (EdgeId | Sequence[EdgeId] | tuple[str, str] | Sequence[tuple[str, str]] | None) – Edge ids
sources (ViewIdentifier | Sequence[ViewIdentifier] | View | Sequence[View] | None) – Retrieve properties from the listed - by reference - views.
include_typing (bool) – Whether to return property type information as part of the result.

Returns

Requested instances.

Return type

InstancesResult

Examples

Retrieve instances by id:

>>> from cognite.client import CogniteClient
>>> c = CogniteClient()
>>> res = c.data_modeling.instances.retrieve(nodes=("mySpace", "myNodeExternalId"),
...                                          edges=("mySpace", "myEdgeExternalId"),
...                                          sources=("mySpace", "myViewExternalId", "myViewVersion")
...                                         )

Retrieve nodes an edges using the built in data class

>>> from cognite.client import CogniteClient
>>> from cognite.client.data_classes.data_modeling import NodeId, EdgeId, ViewId
>>> c = CogniteClient()
>>> res = c.data_modeling.instances.retrieve(NodeId("mySpace", "myNode"),
...                                          EdgeId("mySpace", "myEdge"),
...                                          ViewId("mySpace", "myViewExternalId", "myViewVersion")
...                                         )

Retrieve nodes an edges using the the view object as source

>>> from cognite.client import CogniteClient
>>> from cognite.client.data_classes.data_modeling import NodeId, EdgeId
>>> c = CogniteClient()
>>> res = c.data_modeling.instances.retrieve(NodeId("mySpace", "myNode"),
...                                          EdgeId("mySpace", "myEdge"),
...                                          sources=("myspace", "myView")
...                                         )

List instances

List instances

Parameters

instance_type (Literal["node", "edge"]) – Whether to query for nodes or edges.
include_typing (bool) – Whether to return property type information as part of the result.
sources (ViewIdentifier | Sequence[ViewIdentifier] | View | Sequence[View] | None) – Views to retrieve properties from.
limit (int | None) – Maximum number of instances to return. Defaults to 25. Set to -1, float(“inf”) or None to return all items.
sort (Sequence[InstanceSort | dict] | InstanceSort | dict | None) – How you want the listed instances information ordered.
filter (Filter | dict | None) – Advanced filtering of instances.

Returns

List of requested instances

Return type

NodeList | EdgeList

Examples

List instances and limit to 5:

>>> from cognite.client import CogniteClient
>>> c = CogniteClient()
>>> instance_list = c.data_modeling.instances.list(limit=5)

Iterate over instances:

>>> from cognite.client import CogniteClient
>>> c = CogniteClient()
>>> for instance in c.data_modeling.instances:
...     instance # do something with the instance

Iterate over chunks of instances to reduce memory load:

>>> from cognite.client import CogniteClient
>>> c = CogniteClient()
>>> for instance_list in c.data_modeling.instances(chunk_size=100):
...     instance_list # do something with the instances

Apply instances

InstancesAPI.apply(nodes: NodeApply | Sequence[NodeApply] | None = None, edges: EdgeApply | Sequence[EdgeApply] | None = None, auto_create_start_nodes: bool = False, auto_create_end_nodes: bool = False, auto_create_direct_relations: bool = True, skip_on_version_conflict: bool = False, replace: bool = False) → InstancesApplyResult

Add or update (upsert) instances.

Parameters

nodes (NodeApply | Sequence[NodeApply] | None) – Nodes to apply
edges (EdgeApply | Sequence[EdgeApply] | None) – Edges to apply
auto_create_start_nodes (bool) – Whether to create missing start nodes for edges when ingesting. By default, the start node of an edge must exist before it can be ingested.
auto_create_end_nodes (bool) – Whether to create missing end nodes for edges when ingesting. By default, the end node of an edge must exist before it can be ingested.
auto_create_direct_relations (bool) – Whether to create missing direct relation targets when ingesting.
skip_on_version_conflict (bool) – If existingVersion is specified on any of the nodes/edges in the input, the default behaviour is that the entire ingestion will fail when version conflicts occur. If skipOnVersionConflict is set to true, items with version conflicts will be skipped instead. If no version is specified for nodes/edges, it will do the writing directly.
replace (bool) – How do we behave when a property value exists? Do we replace all matching and existing values with the supplied values (true)? Or should we merge in new values for properties together with the existing values (false)? Note: This setting applies for all nodes or edges specified in the ingestion call.

Returns

Created instance(s)

Return type

InstancesApplyResult

Examples

Create new node without data:

>>> from cognite.client import CogniteClient
>>> from cognite.client.data_classes.data_modeling import EdgeApply, NodeOrEdgeData, NodeApply
>>> c = CogniteClient()
>>> nodes = [NodeApply("mySpace", "myNodeId")]
>>> res = c.data_modeling.instances.apply(nodes)

Create two nodes with data with a one-to-many edge

>>> from cognite.client import CogniteClient
>>> from cognite.client.data_classes.data_modeling import EdgeApply, NodeOrEdgeData, NodeApply, ViewId
>>> actor = NodeApply(
...     space="actors",
...     external_id="arnold_schwarzenegger",
...     sources=[
...         NodeOrEdgeData(
...             ViewId("mySpace", "PersonView", "v1"),
...             {"name": "Arnold Schwarzenegger", "birthYear": 1947}
...         ),
...         NodeOrEdgeData(
...             ViewId("mySpace", "ActorView", "v1"),
...             {"wonOscar": False}
...         )
...     ]
... )
>>> movie = NodeApply(
...     space="movies",
...     external_id="Terminator",
...     sources=[
...         NodeOrEdgeData(
...             ViewId("mySpace", "MovieView", "v1"),
...             {"title": "Terminator", "releaseYear": 1984}
...         )
...     ]
... )
... # This is one-to-many edge, in this case from a person to a movie
>>> actor_to_movie = EdgeApply(
...     space="actors",
...     external_id="relation:arnold_schwarzenegger:terminator",
...     type=("types", "acts-in"),
...     start_node=("actors", "arnold_schwarzenegger"),
...     end_node=("movies", "Terminator"),
... )
>>> res = c.data_modeling.instances.apply([actor, movie], [actor_to_movie])

Create new edge and automatically create end nodes.

>>> from cognite.client import CogniteClient
>>> from cognite.client.data_classes.data_modeling import EdgeApply
>>> c = CogniteClient()
>>> actor_to_movie = EdgeApply(
...     space="actors",
...     external_id="relation:arnold_schwarzenegger:terminator",
...     type=("types", "acts-in"),
...     start_node=("actors", "arnold_schwarzenegger"),
...     end_node=("movies", "Terminator"),
... )
>>> res = c.data_modeling.instances.apply(
...     edges=actor_to_movie,
...     auto_create_start_nodes=True,
...     auto_create_end_nodes=True
... )

Search instances

InstancesAPI.search(view: ViewId, query: str, instance_type: Literal['node'] = 'node', properties: list[str] | None = None, filter: Filter | dict | None = None, limit: int = DEFAULT_LIMIT_READ) → NodeList

InstancesAPI.search(view: ViewId, query: str, instance_type: Literal['edge'], properties: list[str] | None = None, filter: Filter | dict | None = None, limit: int = DEFAULT_LIMIT_READ) → EdgeList

Search instances

Parameters

view (ViewId) – View to search in.
query (str) – Query string that will be parsed and used for search.
instance_type (Literal["node", "edge"]) – Whether to search for nodes or edges.
properties (list[str] | None) – Optional array of properties you want to search through. If you do not specify one or more properties, the service will search all text fields within the view.
filter (Filter | dict | None) – Advanced filtering of instances.
limit (int) – Maximum number of instances to return. Defaults to 25.

Returns

Search result with matching nodes or edges.

Return type

NodeList | EdgeList

Examples

Search for Arnold in the person view in the name property:

>>> from cognite.client import CogniteClient
>>> from cognite.client.data_classes.data_modeling import ViewId
>>> c = CogniteClient()
>>> res = c.data_modeling.instances.search(ViewId("mySpace", "PersonView", "v1"), query="Arnold", properties=["name"])

Search for Quentin in the person view in the name property, but only born before 1970:

>>> from cognite.client import CogniteClient
>>> from cognite.client.data_classes.data_modeling import ViewId
>>> from cognite.client.data_classes import filters
>>> c = CogniteClient()
>>> born_after_1970 = filters.Range(["mySpace", "PersonView/v1", "birthYear"], gt=1970)
>>> res = c.data_modeling.instances.search(ViewId("mySpace", "PersonView", "v1"),
... query="Quentin", properties=["name"], filter=born_after_1970)

Aggregate instances

InstancesAPI.aggregate(view: ViewId, aggregates: MetricAggregation | dict | Sequence[MetricAggregation | dict], instance_type: Literal['node', 'edge'] = 'node', group_by: Sequence[str] | None = None, query: str | None = None, properties: Sequence[str] | None = None, filter: Filter | None = None, limit: int = 25) → InstanceAggregationResultList

Aggregate data across nodes/edges

Parameters

view (ViewId) – View to to aggregate over.
aggregates (MetricAggregation | dict | Sequence[MetricAggregation | dict]) – The properties to aggregate over.
instance_type (Literal["node", "edge"]) – Whether to search for nodes or edges.
group_by (Sequence[str] | None) – The selection of fields to group the results by when doing aggregations. You can specify up to 5 items to group by.
query (str | None) – Query string that will be parsed and used for search.
properties (Sequence[str] | None) – Optional array of properties you want to search through. If you do not specify one or more properties, the service will search all text fields within the view.
filter (Filter | None) – Advanced filtering of instances.
limit (int) – Maximum number of instances to return. Defaults to 25.

Returns

Node or edge aggregation results.

Return type

InstanceAggregationResultList

Examples

Get the average run time in minutes for movies grouped by release year:

>>> from cognite.client import CogniteClient
>>> from cognite.client.data_classes.data_modeling import ViewId, aggregations as aggs
>>> c = CogniteClient()
>>> avg_run_time = aggs.Avg("runTimeMinutes")
>>> view_id = ViewId("mySpace", "PersonView", "v1")
>>> res = c.data_modeling.instances.aggregate(view_id, [avg_run_time], group_by=["releaseYear"])

Query instances

InstancesAPI.query(query: Query) → QueryResult

Advanced query interface for nodes/edges.

The Data Modelling API exposes an advanced query interface. The query interface supports parameterization, recursive edge traversal, chaining of result sets, and granular property selection.

Parameters: query (Query) – Query.
Returns: The resulting nodes and/or edges from the query.
Return type: QueryResult

Examples

Find actors in movies released before 2000 sorted by actor name:

>>> from cognite.client import CogniteClient
>>> from cognite.client.data_classes.data_modeling.query import Query, Select, NodeResultSetExpression, EdgeResultSetExpression, SourceSelector
>>> from cognite.client.data_classes.filters import Range, Equals
>>> from cognite.client.data_classes.data_modeling.ids import ViewId
>>> c = CogniteClient()
>>> movie_id = ViewId("mySpace", "MovieView", "v1")
>>> actor_id = ViewId("mySpace", "ActorView", "v1")
>>> query = Query(
...     with_ = {
...         "movies": NodeResultSetExpression(filter=Range(movie_id.as_property_ref("releaseYear"), lt=2000)),
...         "actors_in_movie": EdgeResultSetExpression(from_="movies", filter=Equals(["edge", "type"], {"space": movie_id.space, "externalId": "Movie.actors"})),
...         "actors": NodeResultSetExpression(from_="actors_in_movie"),
...     },
...     select = {
...         "actors": Select(
...             [SourceSelector(actor_id, ["name"])], sort=[InstanceSort(actor_id.as_property_ref("name"))]),
...     },
... )
>>> res = c.data_modeling.instances.query(query)

Sync instances

InstancesAPI.sync(query: Query) → QueryResult

Subscription to changes for nodes/edges.

Subscribe to changes for nodes and edges in a project, matching a supplied filter.

Parameters: query (Query) – Query.
Returns: The resulting nodes and/or edges from the query.
Return type: QueryResult

Examples

Find actors in movies released before 2000 sorted by actor name:

>>> from cognite.client import CogniteClient
>>> from cognite.client.data_classes.data_modeling.instances import InstanceSort
>>> from cognite.client.data_classes.data_modeling.query import Query, Select, NodeResultSetExpression, EdgeResultSetExpression, SourceSelector
>>> from cognite.client.data_classes.filters import Range, Equals
>>> from cognite.client.data_classes.data_modeling.ids import ViewId
>>> c = CogniteClient()
>>> movie_id = ViewId("mySpace", "MovieView", "v1")
>>> actor_id = ViewId("mySpace", "ActorView", "v1")
>>> query = Query(
...     with_ = {
...         "movies": NodeResultSetExpression(filter=Range(movie_id.as_property_ref("releaseYear"), lt=2000)),
...         "actors_in_movie": EdgeResultSetExpression(from_="movies", filter=Equals(["edge", "type"], {"space": movie_id.space, "externalId": "Movie.actors"})),
...         "actors": NodeResultSetExpression(from_="actors_in_movie"),
...     },
...     select = {
...         "actors": Select(
...             [SourceSelector(actor_id, ["name"])], sort=[InstanceSort(actor_id.as_property_ref("name"))]),
...     },
... )
>>> res = c.data_modeling.instances.sync(query)
>>> # Added a new movie with actors released before 2000
>>> query.cursors = res.cursors
>>> res_new = c.data_modeling.instances.sync(query)

In the last example, the res_new will only contain the actors that have been added with the new movie.

InstancesAPI.subscribe(query: Query, callback: Callable[[QueryResult], None], poll_delay_seconds: float = 30, throttle_seconds: float = 1) → SubscriptionContext

Subscribe to a query and get updates when the result set changes. This invokes the sync() method in a loop in a background thread, and only invokes the callback when there are actual changes to the result set(s).

We do not support chaining result sets when subscribing to a query.

Parameters

query (Query) – The query to subscribe to.
callback (Callable[[QueryResult], None]) – The callback function to call when the result set changes.
poll_delay_seconds (float) – The time to wait between polls when no data is present. Defaults to 30 seconds.
throttle_seconds (float) – The time to wait between polls despite data being present.

Returns

An object that can be used to cancel the subscription.

Return type

SubscriptionContext

Examples

Subscrie to a given query and print the changed data:

>>> from cognite.client import CogniteClient
>>> from cognite.client.data_classes.data_modeling.query import Query, QueryResult, NodeResultSetExpression, Select, SourceSelector
>>> from cognite.client.data_classes.data_modeling import ViewId
>>> from cognite.client.data_classes.filters import Range
>>>
>>> c = CogniteClient()
>>> def just_print_the_result(result: QueryResult) -> None:
...     print(result)
...
>>> view_id = ViewId("someSpace", "someView", "v1")
>>> filter = Range(view_id.as_property_ref("releaseYear"), lt=2000)
>>> query = Query(
...     with_={"movies": NodeResultSetExpression(filter=filter)},
...     select={"movies": Select([SourceSelector(view_id, ["releaseYear"])])}
... )
>>> subscription_context = c.data_modeling.instances.subscribe(query, just_print_the_result)
>>> subscription_context.cancel()

Example on syncing instances to local sqlite

import json
import time
import sqlite3

from cognite.client import CogniteClient
from cognite.client.data_classes.data_modeling.instances import (
    SubscriptionContext,
)
from cognite.client.data_classes.data_modeling.query import (
    QueryResult,
    Query,
    NodeResultSetExpression,
    Select,
)
from cognite.client.data_classes.filters import Equals

c = CogniteClient()


def sqlite_connection(db_name: str) -> sqlite3.Connection:
    return sqlite3.connect(db_name, check_same_thread=False)


def bootstrap_sqlite(db_name: str) -> None:
    with sqlite_connection(db_name) as connection:
        connection.execute(
            """
            CREATE TABLE IF NOT EXISTS instance (
                space TEXT,
                external_id TEXT,
                data JSON,
                PRIMARY KEY(space, external_id)
            )
            """
        )
        connection.execute("CREATE TABLE IF NOT EXISTS cursor (cursor TEXT)")


def sync_space_to_sqlite(
    db_name: str, space_to_sync: str
) -> SubscriptionContext:
    with sqlite_connection(db_name) as connection:
        existing_cursor = connection.execute(
            "SELECT cursor FROM cursor"
        ).fetchone()
        if existing_cursor:
            print("Found existing cursor, using that")

    query = Query(
        with_={
            "nodes": NodeResultSetExpression(
                filter=Equals(property=["node", "space"], value=space_to_sync)
            )
        },
        select={"nodes": Select()},
        cursors={"nodes": existing_cursor[0] if existing_cursor else None},
    )

    def _sync_batch_to_sqlite(result: QueryResult):
        with sqlite_connection(db_name) as connection:
            inserts = []
            deletes = []
            for node in result["nodes"]:
                if node.deleted_time is None:
                    inserts.append(
                        (node.space, node.external_id, json.dumps(node.dump()))
                    )
                else:
                    deletes.append((node.space, node.external_id))
            # Updates must be done in the same transaction as persisting the cursor.
            # A transaction is implicitly started by sqlite here.
            #
            # It is also important that deletes happen first as the same (space, external_id)
            # may appear as several tombstones and then a new instance, which must result in
            # the instance being saved.
            connection.executemany(
                "DELETE FROM instance WHERE space=? AND external_id=?", deletes
            )
            connection.executemany(
                "INSERT INTO instance VALUES (?, ?, ?) ON CONFLICT DO UPDATE SET data=excluded.data",
                inserts,
            )
            connection.execute(
                "INSERT INTO cursor VALUES (?)", [result.cursors["nodes"]]
            )
            connection.commit()
        print(f"Wrote {len(inserts)} nodes and deleted {len(deletes)} nodes")

    return c.data_modeling.instances.subscribe(query, _sync_batch_to_sqlite)


if __name__ == "__main__":
    SQLITE_DB_NAME = "test.db"
    SPACE_TO_SYNC = "mytestspace"
    bootstrap_sqlite(db_name=SQLITE_DB_NAME)
    sync_space_to_sqlite(db_name=SQLITE_DB_NAME, space_to_sync=SPACE_TO_SYNC)
    while True:
        # Keep main thread alive
        time.sleep(10)

Delete instances

Delete one or more instances

Parameters

nodes (NodeId | Sequence[NodeId] | tuple[str, str] | Sequence[tuple[str, str]] | None) – Node ids
edges (EdgeId | Sequence[EdgeId] | tuple[str, str] | Sequence[tuple[str, str]] | None) – Edge ids

Returns

The instance(s) which has been deleted. Empty list if nothing was deleted.

Return type

InstancesDeleteResult

Examples

Delete instances by id:

>>> from cognite.client import CogniteClient
>>> c = CogniteClient()
>>> c.data_modeling.instances.delete(nodes=("mySpace", "myNode"))

Delete nodes and edges using the built in data class

>>> from cognite.client import CogniteClient
>>> from cognite.client.data_classes.data_modeling import NodeId, EdgeId
>>> c = CogniteClient()
>>> c.data_modeling.instances.delete(NodeId('mySpace', 'myNode'), EdgeId('mySpace', 'myEdge'))

Delete all nodes from a NodeList

>>> from cognite.client import CogniteClient
>>> from cognite.client.data_classes.data_modeling import NodeId, EdgeId
>>> c = CogniteClient()
>>> my_view = c.data_modeling.views.retrieve('mySpace', 'myView')
>>> my_nodes = c.data_modeling.instances.list(instance_type='node', sources=my_view, limit=None)
>>> c.data_modeling.instances.delete(nodes=my_nodes.as_ids())

Instances core data classes

class cognite.client.data_classes.data_modeling.instances.Edge(space: str, external_id: str, version: int, type: DirectRelationReference, last_updated_time: int, created_time: int, start_node: DirectRelationReference, end_node: DirectRelationReference, deleted_time: int | None = None, properties: Properties | None = None, **_: Any)

Bases: Instance

An Edge. This is the read version of the edge.

Parameters

space (str) – The workspace for the edge, a unique identifier for the space.
external_id (str) – Combined with the space is the unique identifier of the edge.
version (int) – DMS version.
type (DirectRelationReference) – The type of edge.
last_updated_time (int) – The number of milliseconds since 00:00:00 Thursday, 1 January 1970, Coordinated Universal Time (UTC), minus leap seconds.
created_time (int) – The number of milliseconds since 00:00:00 Thursday, 1 January 1970, Coordinated Universal Time (UTC), minus leap seconds.
start_node (DirectRelationReference) – Reference to the direct relation. The reference consists of a space and an external-id.
end_node (DirectRelationReference) – Reference to the direct relation. The reference consists of a space and an external-id.
deleted_time (int | None) – The number of milliseconds since 00:00:00 Thursday, 1 January 1970, Coordinated Universal Time (UTC), minus leap seconds. Timestamp when the instance was soft deleted. Note that deleted instances are filtered out of query results, but present in sync results
properties (Properties | None) – No description.
**_ (Any) – No description.

as_apply(source: ViewIdentifier | ContainerIdentifier, existing_version: int | None = None) → EdgeApply

This is a convenience function for converting the read to a write edge.

It makes the simplifying assumption that all properties are from the same view. Note that this is not true in general.

Parameters

source (ViewIdentifier | ContainerIdentifier) – The view or container to with all the properties.
existing_version (int | None) – Fail the ingestion request if the node’s version is greater than or equal to this value. If no existingVersion is specified, the ingestion will always overwrite any existing data for the edge (for the specified container or instance). If existingVersion is set to 0, the upsert will behave as an insert, so it will fail the bulk if the item already exists. If skipOnVersionConflict is set on the ingestion request, then the item will be skipped instead of failing the ingestion request.

Returns

A write edge, EdgeApply

Return type

EdgeApply

dump(camel_case: bool = False) → dict[str, Any]

Dump the instance into a json serializable Python data type.

Parameters: camel_case (bool) – Use camelCase for attribute names. Defaults to False.
Returns: A dictionary representation of the instance.
Return type: dict[str, Any]

class cognite.client.data_classes.data_modeling.instances.EdgeApply(space: str, external_id: str, type: DirectRelationReference | tuple[str, str], start_node: DirectRelationReference | tuple[str, str], end_node: DirectRelationReference | tuple[str, str], existing_version: int | None = None, sources: list[NodeOrEdgeData] | None = None)

Bases: InstanceApply

An Edge. This is the write version of the edge.

Parameters

space (str) – The workspace for the edge, a unique identifier for the space.
external_id (str) – Combined with the space is the unique identifier of the edge.
type (DirectRelationReference | tuple[str, str]) – The type of edge.
start_node (DirectRelationReference | tuple[str, str]) – Reference to the direct relation. The reference consists of a space and an external-id.
end_node (DirectRelationReference | tuple[str, str]) – Reference to the direct relation. The reference consists of a space and an external-id.
existing_version (int | None) – Fail the ingestion request if the node’s version is greater than or equal to this value. If no existingVersion is specified, the ingestion will always overwrite any existing data for the edge (for the specified container or edge). If existingVersion is set to 0, the upsert will behave as an insert, so it will fail the bulk if the item already exists. If skipOnVersionConflict is set on the ingestion request, then the item will be skipped instead of failing the ingestion request.
sources (list[NodeOrEdgeData] | None) – List of source properties to write. The properties are from the edge and/or container the container(s) making up this node.

dump(camel_case: bool = False) → dict[str, Any]

Dump the instance into a json serializable Python data type.

Parameters: camel_case (bool) – Use camelCase for attribute names. Defaults to False.
Returns: A dictionary representation of the instance.
Return type: dict[str, Any]

class cognite.client.data_classes.data_modeling.instances.EdgeApplyList(resources: Collection[Any], cognite_client: CogniteClient | None = None)

Bases: CogniteResourceList[EdgeApply]

as_ids() → list[EdgeId]

Convert the list of edges to a list of edge ids.

Returns: A list of edge ids.
Return type: list[EdgeId]

class cognite.client.data_classes.data_modeling.instances.EdgeApplyResult(space: str, external_id: str, version: int, was_modified: bool, last_updated_time: int, created_time: int, **_: Any)

Bases: InstanceApplyResult

An Edge. This represents the update on the edge.

Parameters

space (str) – The workspace for the edge, a unique identifier for the space.
external_id (str) – Combined with the space is the unique identifier of the edge.
version (int) – DMS version.
was_modified (bool) – Whether the edge was modified by the ingestion.
last_updated_time (int) – The number of milliseconds since 00:00:00 Thursday, 1 January 1970, Coordinated Universal Time (UTC), minus leap seconds.
created_time (int) – The number of milliseconds since 00:00:00 Thursday, 1 January 1970, Coordinated Universal Time (UTC), minus leap seconds.
**_ (Any) – No description.

class cognite.client.data_classes.data_modeling.instances.EdgeApplyResultList(resources: Collection[Any], cognite_client: CogniteClient | None = None)

Bases: CogniteResourceList[EdgeApplyResult]

as_ids() → list[EdgeId]

Convert the list of edges to a list of edge ids.

Returns: A list of edge ids.
Return type: list[EdgeId]

class cognite.client.data_classes.data_modeling.instances.EdgeList(resources: Collection[Any], cognite_client: CogniteClient | None = None)

Bases: CogniteResourceList[Edge]

as_ids() → list[EdgeId]

Convert the list of edges to a list of edge ids.

Returns: A list of edge ids.
Return type: list[EdgeId]

class cognite.client.data_classes.data_modeling.instances.EdgeListWithCursor(resources: Collection[Any], cursor: str | None, cognite_client: CogniteClient | None = None): Bases: EdgeList

class cognite.client.data_classes.data_modeling.instances.Instance(space: str, external_id: str, version: int, last_updated_time: int, created_time: int, instance_type: Literal['node', 'edge'] = 'node', deleted_time: int | None = None, properties: Properties | None = None, **_: Any)

Bases: InstanceCore

A node or edge. This is the read version of the instance.

Parameters

space (str) – The workspace for the instance, a unique identifier for the space.
external_id (str) – Combined with the space is the unique identifier of the instance.
version (int) – DMS version.
last_updated_time (int) – The number of milliseconds since 00:00:00 Thursday, 1 January 1970, Coordinated Universal Time (UTC), minus leap seconds.
created_time (int) – The number of milliseconds since 00:00:00 Thursday, 1 January 1970, Coordinated Universal Time (UTC), minus leap seconds.
instance_type (Literal["node", "edge"]) – The type of instance.
deleted_time (int | None) – The number of milliseconds since 00:00:00 Thursday, 1 January 1970, Coordinated Universal Time (UTC), minus leap seconds. Timestamp when the instance was soft deleted. Note that deleted instances are filtered out of query results, but present in sync results
properties (Properties | None) – Properties of the instance.
**_ (Any) – This is used to capture any changes in the API without breaking the SDK.

abstract as_apply(source: ViewIdentifier | ContainerIdentifier, existing_version: int) → InstanceApply: Convert the instance to an apply instance.

dump(camel_case: bool = False) → dict[str, Any]

Dump the instance into a json serializable Python data type.

Parameters: camel_case (bool) – Use camelCase for attribute names. Defaults to False.
Returns: A dictionary representation of the instance.
Return type: dict[str, Any]

class cognite.client.data_classes.data_modeling.instances.InstanceAggregationResult(aggregates: list[AggregatedNumberedValue], group: dict[str, str | int | float | bool])

Bases: DataModelingResource

A node or edge. This represents the update on the instance.

Parameters

aggregates (list[AggregatedNumberedValue]) – List of aggregated values.
group (dict[str, str | int | float | bool]) – The grouping used for the aggregation.

dump(camel_case: bool = False) → dict[str, Any]

Dumps the aggregation results to a dictionary.

Parameters: camel_case (bool) – Whether to convert the keys to camel case.
Returns: A dictionary with the instance results.
Return type: dict[str, Any]

classmethod load(data: dict | str) → InstanceAggregationResult

Loads an instance from a json string or dictionary.

Parameters: data (dict | str) – The json string or dictionary.
Returns: An instance.
Return type: InstanceAggregationResult

class cognite.client.data_classes.data_modeling.instances.InstanceAggregationResultList(resources: Collection[Any], cognite_client: CogniteClient | None = None): Bases: CogniteResourceList[InstanceAggregationResult]

class cognite.client.data_classes.data_modeling.instances.InstanceApply(space: str, external_id: str, instance_type: Literal['node', 'edge'] = 'node', existing_version: int | None = None, sources: list[NodeOrEdgeData] | None = None)

Bases: InstanceCore

A node or edge. This is the write version of the instance.

Parameters

space (str) – The workspace for the instance, a unique identifier for the space.
external_id (str) – Combined with the space is the unique identifier of the instance.
instance_type (Literal["node", "edge"]) – No description.
existing_version (int | None) – Fail the ingestion request if the node’s version is greater than or equal to this value. If no existingVersion is specified, the ingestion will always overwrite any existing data for the edge (for the specified container or instance). If existingVersion is set to 0, the upsert will behave as an insert, so it will fail the bulk if the item already exists. If skipOnVersionConflict is set on the ingestion request, then the item will be skipped instead of failing the ingestion request.
sources (list[NodeOrEdgeData] | None) – List of source properties to write. The properties are from the instance and/or container the container(s) making up this node.

dump(camel_case: bool = False) → dict[str, Any]

Dump the instance into a json serializable Python data type.

Parameters: camel_case (bool) – Use camelCase for attribute names. Defaults to False.
Returns: A dictionary representation of the instance.
Return type: dict[str, Any]

class cognite.client.data_classes.data_modeling.instances.InstanceApplyResult(instance_type: Literal['node', 'edge'], space: str, external_id: str, version: int, was_modified: bool, last_updated_time: int, created_time: int, **_: Any)

Bases: InstanceCore

A node or edge. This represents the update on the instance.

Parameters

instance_type (Literal["node", "edge"]) – The type of instance.
space (str) – The workspace for the instance, a unique identifier for the space.
external_id (str) – Combined with the space is the unique identifier of the instance.
version (int) – DMS version of the instance.
was_modified (bool) – Whether the instance was modified by the ingestion.
last_updated_time (int) – The number of milliseconds since 00:00:00 Thursday, 1 January 1970, Coordinated Universal Time (UTC), minus leap seconds.
created_time (int) – The number of milliseconds since 00:00:00 Thursday, 1 January 1970, Coordinated Universal Time (UTC), minus leap seconds.
**_ (Any) – No description.

class cognite.client.data_classes.data_modeling.instances.InstanceCore(space: str, external_id: str, instance_type: Literal['node', 'edge'])

Bases: DataModelingResource

A node or edge :param space: The workspace for the instance, a unique identifier for the space. :type space: str :param external_id: Combined with the space is the unique identifier of the instance. :type external_id: str :param instance_type: No description. :type instance_type: Literal[“node”, “edge”]

class cognite.client.data_classes.data_modeling.instances.InstancesApply(nodes: NodeApplyList, edges: EdgeApplyList)

Bases: object

This represents the write request of an instance query

Parameters

nodes (NodeApplyList) – A list of nodes.
edges (EdgeApplyList) – A list of edges.

class cognite.client.data_classes.data_modeling.instances.InstancesApplyResult(nodes: NodeApplyResultList, edges: EdgeApplyResultList)

Bases: object

This represents the write result of an instance query

Parameters

nodes (NodeApplyResultList) – A list of nodes.
edges (EdgeApplyResultList) – A list of edges.

class cognite.client.data_classes.data_modeling.instances.InstancesDeleteResult(nodes: list[NodeId], edges: list[EdgeId])

Bases: object

This represents the delete result of an instance query

Parameters

nodes (list[NodeId]) – A list of node ids.
edges (list[EdgeId]) – A list of edge ids.

class cognite.client.data_classes.data_modeling.instances.InstancesResult(nodes: NodeList, edges: EdgeList)

Bases: object

This represents the read result of an instance query

Parameters

nodes (NodeList) – A list of nodes.
edges (EdgeList) – A list of edges.

class cognite.client.data_classes.data_modeling.instances.Node(space: str, external_id: str, version: int, last_updated_time: int, created_time: int, deleted_time: int | None, properties: Properties | None, type: DirectRelationReference | None, **_: Any)

Bases: Instance

A node. This is the read version of the node.

Parameters

space (str) – The workspace for the node, a unique identifier for the space.
external_id (str) – Combined with the space is the unique identifier of the node.
version (int) – DMS version.
last_updated_time (int) – The number of milliseconds since 00:00:00 Thursday, 1 January 1970, Coordinated Universal Time (UTC), minus leap seconds.
created_time (int) – The number of milliseconds since 00:00:00 Thursday, 1 January 1970, Coordinated Universal Time (UTC), minus leap seconds.
deleted_time (int | None) – The number of milliseconds since 00:00:00 Thursday, 1 January 1970, Coordinated Universal Time (UTC), minus leap seconds. Timestamp when the instance was soft deleted. Note that deleted instances are filtered out of query results, but present in sync results
properties (Properties | None) – Properties of the node.
type (DirectRelationReference | None) – Direct relation pointing to the type node.
**_ (Any) – No description.

as_apply(source: ViewIdentifier | ContainerIdentifier, existing_version: int) → NodeApply

This is a convenience function for converting the read to a write node.

It makes the simplifying assumption that all properties are from the same view. Note that this is not true in general.

Parameters

source (ViewIdentifier | ContainerIdentifier) – The view or container to with all the properties.
existing_version (int) – Fail the ingestion request if the node’s version is greater than or equal to this value. If no existingVersion is specified, the ingestion will always overwrite any existing data for the edge (for the specified container or instance). If existingVersion is set to 0, the upsert will behave as an insert, so it will fail the bulk if the item already exists. If skipOnVersionConflict is set on the ingestion request, then the item will be skipped instead of failing the ingestion request.

Returns

A write node, NodeApply

Return type

NodeApply

dump(camel_case: bool = False) → dict[str, Any]

Dump the instance into a json serializable Python data type.

Parameters: camel_case (bool) – Use camelCase for attribute names. Defaults to False.
Returns: A dictionary representation of the instance.
Return type: dict[str, Any]

class cognite.client.data_classes.data_modeling.instances.NodeApply(space: str, external_id: str, existing_version: int | None = None, sources: list[NodeOrEdgeData] | None = None, type: DirectRelationReference | tuple[str, str] | None = None)

Bases: InstanceApply

A node. This is the write version of the node.

Parameters

space (str) – The workspace for the node, a unique identifier for the space.
external_id (str) – Combined with the space is the unique identifier of the node.
existing_version (int | None) – Fail the ingestion request if the node’s version is greater than or equal to this value. If no existingVersion is specified, the ingestion will always overwrite any existing data for the edge (for the specified container or node). If existingVersion is set to 0, the upsert will behave as an insert, so it will fail the bulk if the item already exists. If skipOnVersionConflict is set on the ingestion request, then the item will be skipped instead of failing the ingestion request.
sources (list[NodeOrEdgeData] | None) – List of source properties to write. The properties are from the node and/or container the container(s) making up this node.
type (DirectRelationReference | tuple[str, str] | None) – Direct relation pointing to the type node.

dump(camel_case: bool = False) → dict[str, Any]

Dump the instance into a json serializable Python data type.

Parameters: camel_case (bool) – Use camelCase for attribute names. Defaults to False.
Returns: A dictionary representation of the instance.
Return type: dict[str, Any]

class cognite.client.data_classes.data_modeling.instances.NodeApplyList(resources: Collection[Any], cognite_client: CogniteClient | None = None)

Bases: CogniteResourceList[NodeApply]

as_ids() → list[NodeId]

Convert the list of nodes to a list of node ids.

Returns: A list of node ids.
Return type: list[NodeId]

class cognite.client.data_classes.data_modeling.instances.NodeApplyResult(space: str, external_id: str, version: int, was_modified: bool, last_updated_time: int, created_time: int, **_: Any)

Bases: InstanceApplyResult

A node. This represents the update on the node.

Parameters

space (str) – The workspace for the node, a unique identifier for the space.
external_id (str) – Combined with the space is the unique identifier of the node.
version (int) – DMS version of the node.
was_modified (bool) – Whether the node was modified by the ingestion.
last_updated_time (int) – The number of milliseconds since 00:00:00 Thursday, 1 January 1970, Coordinated Universal Time (UTC), minus leap seconds.
created_time (int) – The number of milliseconds since 00:00:00 Thursday, 1 January 1970, Coordinated Universal Time (UTC), minus leap seconds.
**_ (Any) – No description.

class cognite.client.data_classes.data_modeling.instances.NodeApplyResultList(resources: Collection[Any], cognite_client: CogniteClient | None = None)

Bases: CogniteResourceList[NodeApplyResult]

as_ids() → list[NodeId]

Convert the list of nodes to a list of node ids.

Returns: A list of node ids.
Return type: list[NodeId]

class cognite.client.data_classes.data_modeling.instances.NodeList(resources: Collection[Any], cognite_client: CogniteClient | None = None)

Bases: CogniteResourceList[Node]

as_ids() → list[NodeId]

Convert the list of nodes to a list of node ids.

Returns: A list of node ids.
Return type: list[NodeId]

class cognite.client.data_classes.data_modeling.instances.NodeListWithCursor(resources: Collection[Any], cursor: str | None, cognite_client: CogniteClient | None = None): Bases: NodeList

class cognite.client.data_classes.data_modeling.instances.NodeOrEdgeData(source: ContainerId | ViewId, properties: Mapping[str, PropertyValue])

Bases: object

This represents the data values of a node or edge.

Parameters

source (ContainerId | ViewId) – The container or view the node or edge property is in
properties (Mapping[str, PropertyValue]) – The properties of the node or edge.

class cognite.client.data_classes.data_modeling.instances.Properties(properties: MutableMapping[ViewId, MutableMapping[str, Union[str, int, float, bool, dict, List[str], List[int], List[float], List[bool], List[dict], NodeId, DirectRelationReference]]])

Bases: MutableMapping[Union[ViewId, Tuple[str, str], Tuple[str, str, str]], MutableMapping[str, Union[str, int, float, bool, dict, List[str], List[int], List[float], List[bool], List[dict], NodeId, DirectRelationReference]]]

get(view: Union[ViewId, Tuple[str, str], Tuple[str, str, str]]) → MutableMapping[PropertyIdentifier, PropertyValue] | None
get(view: Union[ViewId, Tuple[str, str], Tuple[str, str, str]], default: MutableMapping[PropertyIdentifier, PropertyValue] | _T) → MutableMapping[PropertyIdentifier, PropertyValue] | _T: D.get(k[,d]) -> D[k] if k in D, else d. d defaults to None.

items() → a set-like object providing a view on D's items

keys() → a set-like object providing a view on D's keys

values() → an object providing a view on D's values

class cognite.client.data_classes.data_modeling.instances.SubscriptionContext(last_successful_sync: 'datetime | None' = None, last_successful_callback: 'datetime | None' = None, _canceled: 'bool' = False, _thread: 'threading.Thread | None' = None): Bases: object

Instances query data classes

class cognite.client.data_classes.data_modeling.query.EdgeResultSetExpression(from_: str | None = None, max_distance: int | None = None, direction: Literal['outwards', 'inwards'] = 'outwards', filter: Filter | None = None, node_filter: Filter | None = None, termination_filter: Filter | None = None, limit_each: int | None = None, sort: list[InstanceSort] | None = None, post_sort: list[InstanceSort] | None = None, limit: int | None = None): Bases: ResultSetExpression

class cognite.client.data_classes.data_modeling.query.NodeResultSetExpression(from_: str | None = None, filter: Filter | None = None, sort: list[InstanceSort] | None = None, limit: int | None = None, through: list[str] | tuple[str, str, str] | None = None): Bases: ResultSetExpression

class cognite.client.data_classes.data_modeling.query.Query(with_: dict[str, ResultSetExpression], select: dict[str, Select], parameters: dict[str, PropertyValue] | None = None, cursors: Mapping[str, str | None] | None = None)

Bases: object

Query allows you to do advanced queries on the data model.

Parameters

with_ (dict[str, ResultSetExpression]) – A dictionary of result set expressions to use in the query. The keys are used to reference the result set expressions in the select and parameters.
select (dict[str, Select]) – A dictionary of select expressions to use in the query. The keys must match the keys in the with_ dictionary. The select expressions define which properties to include in the result set.
parameters (dict[str, PropertyValue] | None) – Values in filters can be parameterised. Parameters are provided as part of the query object, and referenced in the filter itself.
cursors (Mapping[str, str | None] | None) – A dictionary of cursors to use in the query. These are for pagination purposes, for example, in the sync endpoint.

class cognite.client.data_classes.data_modeling.query.ResultSetExpression(from_: str | None, filter: Filter | None, limit: int | None, sort: list[InstanceSort] | None): Bases: ABC

class cognite.client.data_classes.data_modeling.query.Select(sources: 'list[SourceSelector]' = <factory>, sort: 'list[InstanceSort]' = <factory>, limit: 'int | None' = None): Bases: object

class cognite.client.data_classes.data_modeling.query.SourceSelector(source: 'ViewId', properties: 'list[str]'): Bases: object

Data Modeling Filtering data classes

final class cognite.client.data_classes.filters.And(*filters: Filter): Bases: CompoundFilter

class cognite.client.data_classes.filters.CompoundFilter(*filters: Filter): Bases: Filter

final class cognite.client.data_classes.filters.ContainsAll(property: Union[str, Tuple[str, ...], List[str], EnumProperty], values: Union[Sequence[Union[str, float, bool, Sequence, Mapping[str, Any], Label]], PropertyReferenceValue, ParameterValue]): Bases: FilterWithPropertyAndValueList

final class cognite.client.data_classes.filters.ContainsAny(property: Union[str, Tuple[str, ...], List[str], EnumProperty], values: Union[Sequence[Union[str, float, bool, Sequence, Mapping[str, Any], Label]], PropertyReferenceValue, ParameterValue]): Bases: FilterWithPropertyAndValueList

final class cognite.client.data_classes.filters.Equals(property: Union[str, Tuple[str, ...], List[str], EnumProperty], value: Union[str, float, bool, Sequence, Mapping[str, Any], Label, PropertyReferenceValue, ParameterValue]): Bases: FilterWithPropertyAndValue

final class cognite.client.data_classes.filters.Exists(property: Union[str, Tuple[str, ...], List[str], EnumProperty]): Bases: FilterWithProperty

class cognite.client.data_classes.filters.Filter: Bases: ABC

class cognite.client.data_classes.filters.FilterWithProperty(property: Union[str, Tuple[str, ...], List[str], EnumProperty]): Bases: Filter

class cognite.client.data_classes.filters.FilterWithPropertyAndValue(property: Union[str, Tuple[str, ...], List[str], EnumProperty], value: Union[str, float, bool, Sequence, Mapping[str, Any], Label, PropertyReferenceValue, ParameterValue]): Bases: FilterWithProperty

class cognite.client.data_classes.filters.FilterWithPropertyAndValueList(property: Union[str, Tuple[str, ...], List[str], EnumProperty], values: Union[Sequence[Union[str, float, bool, Sequence, Mapping[str, Any], Label]], PropertyReferenceValue, ParameterValue]): Bases: FilterWithProperty

class cognite.client.data_classes.filters.GeoJSON(property: Union[str, Tuple[str, ...], List[str], EnumProperty], geometry: Geometry): Bases: FilterWithProperty, ABC

final class cognite.client.data_classes.filters.GeoJSONDisjoint(property: Union[str, Tuple[str, ...], List[str], EnumProperty], geometry: Geometry): Bases: GeoJSON

final class cognite.client.data_classes.filters.GeoJSONIntersects(property: Union[str, Tuple[str, ...], List[str], EnumProperty], geometry: Geometry): Bases: GeoJSON

final class cognite.client.data_classes.filters.GeoJSONWithin(property: Union[str, Tuple[str, ...], List[str], EnumProperty], geometry: Geometry): Bases: GeoJSON

final class cognite.client.data_classes.filters.HasData(containers: Sequence[tuple[str, str] | ContainerId] | None = None, views: Sequence[tuple[str, str, str] | ViewId] | None = None): Bases: Filter

final class cognite.client.data_classes.filters.In(property: Union[str, Tuple[str, ...], List[str], EnumProperty], values: Union[Sequence[Union[str, float, bool, Sequence, Mapping[str, Any], Label]], PropertyReferenceValue, ParameterValue]): Bases: FilterWithPropertyAndValueList

final class cognite.client.data_classes.filters.InAssetSubtree(property: Union[str, Tuple[str, ...], List[str], EnumProperty], value: Union[str, float, bool, Sequence, Mapping[str, Any], Label, PropertyReferenceValue, ParameterValue]): Bases: FilterWithPropertyAndValue

final class cognite.client.data_classes.filters.MatchAll: Bases: Filter

final class cognite.client.data_classes.filters.Nested(scope: Union[str, Tuple[str, ...], List[str], EnumProperty], filter: Filter): Bases: Filter

final class cognite.client.data_classes.filters.Not(filter: Filter): Bases: CompoundFilter

final class cognite.client.data_classes.filters.Or(*filters: Filter): Bases: CompoundFilter

final class cognite.client.data_classes.filters.Overlaps(start_property: PropertyReference, end_property: PropertyReference, gt: FilterValue | None = None, gte: FilterValue | None = None, lt: FilterValue | None = None, lte: FilterValue | None = None): Bases: Filter

class cognite.client.data_classes.filters.ParameterValue(parameter: 'str'): Bases: object

final class cognite.client.data_classes.filters.Prefix(property: Union[str, Tuple[str, ...], List[str], EnumProperty], value: Union[str, float, bool, Sequence, Mapping[str, Any], Label, PropertyReferenceValue, ParameterValue]): Bases: FilterWithPropertyAndValue

class cognite.client.data_classes.filters.PropertyReferenceValue(property: 'PropertyReference'): Bases: object

final class cognite.client.data_classes.filters.Range(property: PropertyReference, gt: FilterValue | None = None, gte: FilterValue | None = None, lt: FilterValue | None = None, lte: FilterValue | None = None): Bases: FilterWithProperty

final class cognite.client.data_classes.filters.Search(property: Union[str, Tuple[str, ...], List[str], EnumProperty], value: Union[str, float, bool, Sequence, Mapping[str, Any], Label, PropertyReferenceValue, ParameterValue]): Bases: FilterWithPropertyAndValue

Data Modeling ID data classes

class cognite.client.data_classes.data_modeling.ids.AbstractDataclass(*args: 'Any', **kwargs: 'Any'): Bases: ABC

class cognite.client.data_classes.data_modeling.ids.ContainerId(*args: 'Any', **kwargs: 'Any'): Bases: DataModelingId

class cognite.client.data_classes.data_modeling.ids.DataModelId(*args: 'Any', **kwargs: 'Any'): Bases: VersionedDataModelingId

class cognite.client.data_classes.data_modeling.ids.DataModelingId(*args: 'Any', **kwargs: 'Any'): Bases: AbstractDataclass

class cognite.client.data_classes.data_modeling.ids.EdgeId(space: 'str', external_id: 'str'): Bases: InstanceId

class cognite.client.data_classes.data_modeling.ids.IdLike(*args, **kwargs): Bases: Protocol

class cognite.client.data_classes.data_modeling.ids.InstanceId(space: 'str', external_id: 'str'): Bases: object

class cognite.client.data_classes.data_modeling.ids.NodeId(space: 'str', external_id: 'str'): Bases: InstanceId

class cognite.client.data_classes.data_modeling.ids.VersionedDataModelingId(*args: 'Any', **kwargs: 'Any'): Bases: AbstractDataclass

class cognite.client.data_classes.data_modeling.ids.VersionedIdLike(*args, **kwargs): Bases: IdLike, Protocol

class cognite.client.data_classes.data_modeling.ids.ViewId(*args: 'Any', **kwargs: 'Any'): Bases: VersionedDataModelingId

GraphQL

Apply DML

DataModelingGraphQLAPI.apply_dml(id: DataModelIdentifier, dml: str, name: str | None = None, description: str | None = None, previous_version: str | None = None) → DMLApplyResult

Apply the DML for a given data model.

Parameters

id (DataModelIdentifier) – The data model to apply DML to.
dml (str) – The DML to apply.
name (str | None) – The name of the data model.
description (str | None) – The description of the data model.
previous_version (str | None) – The previous version of the data model. Specify to reuse view versions from previous data model version.

Returns

The id of the updated data model.

Return type

DMLApplyResult

Examples

Apply DML:

>>> from cognite.client import CogniteClient
>>> c = CogniteClient()
>>> res = c.data_modeling.graphql.apply_dml(
...     id=("mySpaceExternalId", "myModelExternalId", "1"),
...     dml="type MyType { id: String! }",
...     name="My model name",
...     description="My model description"
... )