Instances

`AsyncCogniteClient.data_modeling.instances.aggregate`(...)	Aggregate data across nodes/edges.
`AsyncCogniteClient.data_modeling.instances.apply`([...])	Add or update (upsert) instances.
`AsyncCogniteClient.data_modeling.instances.delete`([...])	Delete one or more instances.
`AsyncCogniteClient.data_modeling.instances.histogram`(...)	Produces histograms for nodes/edges.
`AsyncCogniteClient.data_modeling.instances.inspect`([...])	Reverse lookup for instances.
`AsyncCogniteClient.data_modeling.instances.list`([...])	List instances.
`AsyncCogniteClient.data_modeling.instances.query`(query)	Advanced query interface for nodes/edges.
`AsyncCogniteClient.data_modeling.instances.retrieve`([...])	Retrieve one or more instance by id(s).
`AsyncCogniteClient.data_modeling.instances.retrieve_edges`(edges)	Retrieve one or more edges by id(s).
`AsyncCogniteClient.data_modeling.instances.retrieve_nodes`(nodes)	Retrieve one or more nodes by id(s).
`AsyncCogniteClient.data_modeling.instances.search`(view)	Search instances.
`AsyncCogniteClient.data_modeling.instances.subscribe`(...)	Subscribe to a query and get updates when the result set changes.
`AsyncCogniteClient.data_modeling.instances.sync`(query)	Subscription to changes for nodes/edges.
`AsyncCogniteClient.data_modeling.instances.sync_with_file_cache`(...)	Create a managed sync session with persistent backup to a CDF file.

Example: Syncing instances to a local SQLite database

The following example demonstrates how to use the subscribe method to create a live, local replica of instances from a specific space in a SQLite database.

import asyncio
import json
import sqlite3
from typing import Optional

from cognite.client import AsyncCogniteClient
from cognite.client.config import ClientConfig
from cognite.client.data_classes.data_modeling import (
    QueryResult,
    QuerySync,
    NodeResultSetExpressionSync,
    SelectSync,
    SubscriptionContext,
)
from cognite.client.data_classes.filters import Equals


def sqlite_connection(db_name: str) -> sqlite3.Connection:
    return sqlite3.connect(db_name)


def bootstrap_sqlite(db_name: str) -> None:
    """Sets up the initial database schema if it doesn't exist."""
    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 (
                space TEXT PRIMARY KEY,
                cursor TEXT
            )
            """
        )
        connection.commit()


async def sync_space_to_sqlite(
    async_client: AsyncCogniteClient, db_name: str, space_to_sync: str
) -> SubscriptionContext:
    """
    Sets up and starts a subscription to sync a space to a local SQLite database.
    """
    # 1. Read the last known cursor from the database.
    #    This is a blocking call, so we run it in a thread.
    def _get_cursor() -> Optional[str]:
        with sqlite_connection(db_name) as connection:
            result = connection.execute(
                "SELECT cursor FROM cursor WHERE space = ?", (space_to_sync,)
            ).fetchone()
            if result:
                print(f"Found existing cursor for space {space_to_sync!r}")
                return result[0]
            return None

    existing_cursor = await asyncio.to_thread(_get_cursor)

    query = QuerySync(
        with_={
            "nodes": NodeResultSetExpressionSync(
                filter=Equals(property=["node", "space"], value=space_to_sync)
            )
        },
        select={"nodes": SelectSync()},
        cursors={"nodes": existing_cursor},
    )

    # 2. Define the callback that will process each batch of results.
    #    The callback itself does not have to be async, but it is preferable.
    async def _sync_batch_to_sqlite(result: QueryResult) -> None:
        def _write_to_db() -> tuple[int, int]:
            # 3. Prepare all data in memory before opening the database
            #    connection to minimize lock time.
            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))

            # 4. Perform all database operations within a single transaction
            #    to ensure data consistency.
            with sqlite_connection(db_name) as connection:
                # Note: Deletions must be processed before insertions. This ensures
                # that we don't lose an instance that has been deleted and re-created
                # in the same sync batch.
                connection.executemany(
                    "DELETE FROM instance WHERE space=? AND external_id=?", deletes
                )
                connection.executemany(
                    "INSERT INTO instance (space, external_id, data) VALUES (?, ?, ?) "
                    "ON CONFLICT(space, external_id) DO UPDATE SET data=excluded.data",
                    inserts,
                )
                # Finally, persist the cursor for this space.
                connection.execute(
                    "INSERT INTO cursor (space, cursor) VALUES (?, ?) "
                    "ON CONFLICT(space) DO UPDATE SET cursor=excluded.cursor",
                    (space_to_sync, result.cursors["nodes"]),
                )
                connection.commit()

            return len(inserts), len(deletes)

        # 5. Run the blocking database write operation in a separate thread.
        inserts, deletes = await asyncio.to_thread(_write_to_db)
        print(f"Wrote {inserts} nodes and deleted {deletes} nodes for space {space_to_sync!r}")

    # 6. Start the subscription and return the SubscriptionContext.
    return await async_client.data_modeling.instances.subscribe(query, _sync_batch_to_sqlite)


async def main():
    async_client = AsyncCogniteClient(ClientConfig(...))

    SQLITE_DB_NAME = "my_instances.db"
    SPACE_TO_SYNC = "my-awesome-space"

    bootstrap_sqlite(db_name=SQLITE_DB_NAME)

    print(f"Starting subscription for space: {SPACE_TO_SYNC!r}...")
    subscription = await sync_space_to_sqlite(
        async_client, db_name=SQLITE_DB_NAME, space_to_sync=SPACE_TO_SYNC
    )
    print("Subscription is live. Press Ctrl+C (or Cmd+C) to stop.")
    try:
        # Keep the application alive to allow the background subscription to run.
        while True:
            await asyncio.sleep(10)

    except asyncio.CancelledError:
        print("Main task cancelled.")
    finally:
        # Ensure we clean up and cancel the subscription task on exit.
        print("Stopping subscription...")
        subscription.cancel()
        # Give the task a moment to shut down gracefully
        await asyncio.sleep(1)
        print("Subscription stopped gracefully.")


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        pass

Instances core data classes

class cognite.client.data_classes.data_modeling.instances.DataModelingInstancesList( resources: Sequence[T_CogniteResource], )

Bases: WriteableCogniteResourceList[T_WriteClass, T_Instance], ABC

extend( other: Iterable[Any], ) → None: S.extend(iterable) – extend sequence by appending elements from the iterable

get( instance_id: InstanceId | tuple[str, str] | None = None, external_id: str | None = None, ) → T_Instance | None

Get an instance from this list by instance ID.

Parameters:

instance_id (InstanceId | tuple[str, str] | None) – The instance ID to get. A tuple on the form (space, external_id) is also accepted.
external_id (str | None) – The external ID of the instance to return. Will raise ValueError when ambiguous (in presence of multiple spaces).

Returns:

The requested instance if present, else None

Return type:

T_Instance | None

to_pandas( camel_case: bool = False, convert_timestamps: bool = True, expand_properties: bool = True, remove_property_prefix: bool = True, ) → pd.DataFrame

Convert the instance into a pandas DataFrame. Note that if the properties column is expanded and there are keys in the metadata that already exist in the DataFrame, then an error will be raised by pandas during joining.

Parameters:

camel_case (bool) – Convert column names to camel case (e.g. externalId instead of external_id). Does not apply to properties.
convert_timestamps (bool) – Convert known columns storing CDF timestamps (milliseconds since epoch) to datetime. Does not affect properties.
expand_properties (bool) – Expand the properties into separate columns.
remove_property_prefix (bool) – Attempt to remove the view ID prefix from columns names of expanded properties. Requires data to be from a single view and that all property names do not conflict with base properties (e.g. ‘space’ or ‘type’). In such cases, a warning is issued and the prefix is kept.

Returns:

The Cognite resource as a dataframe.

Return type:

pd.DataFrame

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

Bases: Instance[EdgeApply]

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) – Current version of the edge.
type (DirectRelationReference | tuple[str, str]) – 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 | 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.
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.

as_apply() → EdgeApply

This is a convenience method for converting from the read version of the Edge to the write version (EdgeApply).

Warning

Properties can be read-only (e.g. if using auto-increment) and then the converted write edge will fail on ingestion.

Returns:: A write edge, EdgeApply, with all properties (even read-only) copied over.
Return type:: EdgeApply

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.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[EdgeApply]

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.

as_write() → EdgeApply: Returns this EdgeApply instance

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.data_modeling.instances.EdgeApplyList( resources: Sequence[T_CogniteResource], )

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, )

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) – Current version of the edge.
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.

class cognite.client.data_classes.data_modeling.instances.EdgeApplyResultList( resources: Sequence[T_CogniteResource], )

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: Sequence[T_Edge], typing: TypeInformation | None = None, debug: DebugInfo | None = None, )

Bases: DataModelingInstancesList[EdgeApply, T_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]

as_write() → EdgeApplyList: Returns this EdgeList as a EdgeApplyList

dump_raw(camel_case: bool = True) → dict[str, Any]

This method dumps the list with extra information in addition to the items.

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

class cognite.client.data_classes.data_modeling.instances.EdgeListWithCursor( resources: Sequence[T_Edge], cursor: str | None, typing: TypeInformation | None = None, debug: DebugInfo | None = None, )

Bases: EdgeList

extend( other: EdgeListWithCursor, ) → None: S.extend(iterable) – extend sequence by appending elements from the iterable

class cognite.client.data_classes.data_modeling.instances.InspectOperation: Bases: ABC

class cognite.client.data_classes.data_modeling.instances.InspectionResults( involved_views: list[ViewId] | None, involved_containers: list[ContainerId] | None, )

Bases: CogniteResource

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.data_modeling.instances.Instance( space: str, external_id: str, version: int, last_updated_time: int, created_time: int, instance_type: Literal['node', 'edge'], deleted_time: int | None, properties: Properties | None, )

Bases: WritableInstanceCore[T_CogniteResource], ABC

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) – Current version of the instance.
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.

abstract as_apply() → InstanceApply: Convert the instance to an apply instance.

drop_source( source: ViewId, ) → None: Remove a source with its corresponding properties from the instance and attempts to reset the instance to be a ‘singular-source instance’ if possible. This restores the ability to use quick property access, e.g. instance[“some_prop”] and makes to_pandas() remove the view ID prefix from column names.

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]

to_pandas( ignore: list[str] | None = None, camel_case: bool = False, convert_timestamps: bool = True, expand_properties: bool = True, remove_property_prefix: bool = True, ) → pd.DataFrame

Convert the instance into a pandas DataFrame.

Parameters:

ignore (list[str] | None) – List of row keys to skip when converting to a data frame. Is applied before expansions.
camel_case (bool) – Convert attribute names to camel case (e.g. externalId instead of external_id). Does not affect properties if expanded.
convert_timestamps (bool) – Convert known attributes storing CDF timestamps (milliseconds since epoch) to datetime. Does not affect properties.
expand_properties (bool) – Expand the properties into separate rows.
remove_property_prefix (bool) – Attempt to remove the view ID prefix from row names of expanded properties (in index). Requires data to be from a single view and that all property names do not conflict with base properties (e.g. ‘space’ or ‘type’). In such cases, a warning is issued and the prefix is kept.

Returns:

The dataframe.

Return type:

pd.DataFrame

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

Bases: DataModelingResource

Represents instances aggregation results.

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 = True, ) → 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]

class cognite.client.data_classes.data_modeling.instances.InstanceAggregationResultList( resources: Sequence[T_CogniteResource], ): 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: WritableInstanceCore[T_CogniteResource], ABC

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']) – The type of instance.
existing_version (int | None) – Fail the ingestion request if the instance’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 instance (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 instance already exists. If skipOnVersionConflict is set on the ingestion request, then the instance 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 = 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.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, )

Bases: InstanceCore, ABC

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.

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

Bases: DataModelingResource, ABC

A node or edge

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']) – The type of instance.

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

Bases: CogniteResource

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.data_modeling.instances.InstanceInspectResultList( resources: Sequence[T_CogniteResource], ): Bases: CogniteResourceList[InstanceInspectResult]

class cognite.client.data_classes.data_modeling.instances.InstanceInspectResults( nodes: 'InstanceInspectResultList', edges: 'InstanceInspectResultList', ): Bases: object

class cognite.client.data_classes.data_modeling.instances.InstanceSort( property: list[str] | tuple[str, ...], direction: Literal['ascending', 'descending'] = 'ascending', nulls_first: bool | None = None, ): Bases: DataModelingSort

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[T_Node], edges: EdgeList[T_Edge], )

Bases: Generic[T_Node, T_Edge]

This represents the read result of an instance query

Parameters:

nodes (NodeList[T_Node]) – A list of nodes.
edges (EdgeList[T_Edge]) – A list of edges.

class cognite.client.data_classes.data_modeling.instances.InvolvedContainers: Bases: InspectOperation

class cognite.client.data_classes.data_modeling.instances.InvolvedViews(all_versions: 'bool' = False): Bases: InspectOperation

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 | tuple[str, str] | None, )

Bases: Instance[NodeApply]

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) – Current version of the node.
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 | tuple[str, str] | None) – Direct relation pointing to the type node.

as_apply() → NodeApply

This is a convenience method for converting from the read version of the Node to the write version (NodeApply).

Warning

Properties can be read-only and then the converted write node will fail on ingestion. Examples are auto-increment properties, or system-controlled ones like path or root (CogniteAsset), or isUploaded (CogniteFile).

Returns:: A write node, NodeApply, with all properties (even read-only) copied over.
Return type:: NodeApply

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.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[NodeApply]

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.

as_write() → Self: Returns this NodeApply instance

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.data_modeling.instances.NodeApplyList( resources: Sequence[T_CogniteResource], )

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, )

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) – Current 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.

class cognite.client.data_classes.data_modeling.instances.NodeApplyResultList( resources: Sequence[T_CogniteResource], )

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: Sequence[T_Node], typing: TypeInformation | None = None, debug: DebugInfo | None = None, )

Bases: DataModelingInstancesList[NodeApply, T_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]

as_write() → NodeApplyList: Returns this NodeList as a NodeApplyList

dump_raw(camel_case: bool = True) → dict[str, Any]

This method dumps the list with extra information in addition to the items.

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

class cognite.client.data_classes.data_modeling.instances.NodeListWithCursor( resources: Sequence[T_Node], cursor: str | None, typing: TypeInformation | None = None, debug: DebugInfo | None = None, )

Bases: NodeList[T_Node]

extend( other: NodeListWithCursor, ) → None: S.extend(iterable) – extend sequence by appending elements from the iterable

Bases: CogniteResource

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, PropertyValueWrite]) – The properties of the node or edge.

dump(camel_case: bool = True) → dict

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]

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.PropertyOptions(identifier: str | None = None)

Bases: object

This is a descriptor class for instance properties in a typed class.

It is used when you have a property that has a different name in the Data Model compared to the name in the Python class.

Parameters:: identifier (str | None) – The name of the property in the Data Model. Defaults to the name of the property in the Python class.

class cognite.client.data_classes.data_modeling.instances.TargetUnit(property: 'str', unit: 'UnitReference | UnitSystemReference')

Bases: CogniteResource

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.data_modeling.instances.TypeInformation( data: dict[str, dict[str, dict[str, TypePropertyDefinition]]] | None = None, )

Bases: UserDict, CogniteResource

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]

to_pandas() → pd.DataFrame

Convert the instance into a pandas DataFrame.

Parameters:

expand_metadata (bool) – Expand the metadata into separate rows (default: False).
metadata_prefix (str) – Prefix to use for the metadata rows, if expanded.
ignore (list[str] | None) – List of row keys to skip when converting to a data frame. Is applied before expansions.
camel_case (bool) – Convert attribute names to camel case (e.g. externalId instead of external_id). Does not affect custom data like metadata if expanded.
convert_timestamps (bool) – Convert known attributes storing CDF timestamps (milliseconds since epoch) to datetime. Does not affect custom data like metadata.

Returns:

The dataframe.

Return type:

pandas.DataFrame

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

Bases: CogniteResource

dump( camel_case: bool = True, return_flat_dict: 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 True.
Returns:: A dictionary representation of the instance.
Return type:: dict[str, Any]

class cognite.client.data_classes.data_modeling.instances.TypedEdge( space: str, external_id: str, version: int, type: DirectRelationReference | tuple[str, str], last_updated_time: int, created_time: int, start_node: DirectRelationReference | tuple[str, str], end_node: DirectRelationReference | tuple[str, str], deleted_time: int | None, ): Bases: Edge, TypedInstance

class cognite.client.data_classes.data_modeling.instances.TypedEdgeApply( 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, ): Bases: EdgeApply, TypedInstance

class cognite.client.data_classes.data_modeling.instances.TypedInstance: Bases: ABC

class cognite.client.data_classes.data_modeling.instances.TypedNode( space: str, external_id: str, version: int, last_updated_time: int, created_time: int, deleted_time: int | None, type: DirectRelationReference | tuple[str, str] | None, ): Bases: Node, TypedInstance

class cognite.client.data_classes.data_modeling.instances.TypedNodeApply( space: str, external_id: str, existing_version: int | None = None, type: ~cognite.client.data_classes.data_modeling.data_types.DirectRelationReference | tuple[str, str] | None | ~cognite.client._constants.Omitted = <Omitted parameter>, )

Bases: NodeApply, TypedInstance

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.data_modeling.instances.WritableInstanceCore( space: str, external_id: str, instance_type: Literal['node', 'edge'], ): Bases: WritableDataModelingResource[T_CogniteResource], ABC

Instances query data classes

class cognite.client.data_classes.data_modeling.query.EdgeResultSetExpression( from_: str | None = None, filter: ~cognite.client.data_classes.filters.Filter | None = None, limit: int | None = None, sort: list[~cognite.client.data_classes.data_modeling.instances.InstanceSort] = <factory>, direction: ~typing.Literal['outwards', 'inwards'] = 'outwards', chain_to: ~typing.Literal['destination', 'source'] = 'destination', max_distance: int | None = None, node_filter: ~cognite.client.data_classes.filters.Filter | None = None, termination_filter: ~cognite.client.data_classes.filters.Filter | None = None, limit_each: int | None = None, post_sort: list[~cognite.client.data_classes.data_modeling.instances.InstanceSort] = <factory>, )

Bases: NodeOrEdgeResultSetExpression

Describes how to query for edges in the data model.

Parameters:

from (str | None) – Chain your result expression from this edge.
filter (Filter | None) – Filter the result set based on this filter.
limit (int | None) – Limit the result set to this number of instances.
sort (list[InstanceSort]) – Sort the result set based on this list of sort criteria.
direction (Literal['outwards', 'inwards']) – The direction to use when traversing.
chain_to (Literal['destination', 'source']) – Control which side of the edge to chain to. The chain_to option is only applicable if the result rexpression referenced in from contains edges. source will chain to start if you’re following edges outwards i.e direction=outwards. If you’re following edges inwards i.e direction=inwards, it will chain to end. destination (default) will chain to end if you’re following edges outwards i.e direction=outwards. If you’re following edges inwards i.e, direction=inwards, it will chain to start.
max_distance (int | None) – The largest - max - number of levels to traverse.
node_filter (Filter | None) – Filter the result set based on this filter.
termination_filter (Filter | None) – Filter the result set based on this filter.
limit_each (int | None) – Limit the number of returned edges for each of the source nodes in the result set. The indicated uniform limit applies to the result set from the referenced from. limitEach only has meaning when you also specify maxDistance=1 and from.
post_sort (list[InstanceSort]) – Sort the result set based on this list of sort criteria.

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.data_modeling.query.EdgeResultSetExpressionSync( from_: str | None = None, filter: ~cognite.client.data_classes.filters.Filter | None = None, limit: int | None = None, direction: ~typing.Literal['outwards', 'inwards'] = 'outwards', chain_to: ~typing.Literal['destination', 'source'] = 'destination', skip_already_deleted: bool = True, sync_mode: ~typing.Literal['one_phase', 'two_phase', 'no_backfill'] = 'two_phase', backfill_sort: list[~cognite.client.data_classes.data_modeling.instances.InstanceSort] = <factory>, max_distance: int | None = None, node_filter: ~cognite.client.data_classes.filters.Filter | None = None, termination_filter: ~cognite.client.data_classes.filters.Filter | None = None, )

Bases: ResultSetExpressionSync

Describes how to query for edges in the data model.

Parameters:

from (str | None) – Chain your result expression from this edge.
filter (Filter | None) – Filter the result set based on this filter.
limit (int | None) – Limit the result set to this number of instances.
direction (Literal['outwards', 'inwards']) – The direction to use when traversing.
chain_to (Literal['destination', 'source']) – Control which side of the edge to chain to. The chain_to option is only applicable if the result rexpression referenced in from contains edges. source will chain to start if you’re following edges outwards i.e direction=outwards. If you’re following edges inwards i.e direction=inwards, it will chain to end. destination (default) will chain to end if you’re following edges outwards i.e direction=outwards. If you’re following edges inwards i.e, direction=inwards, it will chain to start.
skip_already_deleted (bool) – If set to False, the API will return instances that have been soft deleted before sync was initiated. Soft deletes that happen after the sync is initiated and a cursor generated, are always included in the result. Soft deleted instances are identified by having deletedTime set.
sync_mode (SyncMode) – Specify whether to sync instances in a single phase; in a backfill phase followed by live updates, or without any backfill. Defaults to “two_phase”, which is recommended for queries with custom filters. Note that the backfill phase’s filters and sort must be backed by a cursorable index. Use “one_phase” when fetching all instances (or all in specific spaces) for better throughput.
backfill_sort (list[InstanceSort]) – Sort the result set during the backfill phase of a two phase sync. Only valid with sync_mode = “two_phase”. The sort must be backed by a cursorable index.
max_distance (int | None) – The largest - max - number of levels to traverse.
node_filter (Filter | None) – Filter the result set based on this filter.
termination_filter (Filter | None) – Filter the result set based on this filter.

Examples

Import and instantiate:

>>> from cognite.client.data_classes.data_modeling import (
...     EdgeResultSetExpressionSync,
... )
>>> from cognite.client.data_classes.filters import Equals
>>> expr = EdgeResultSetExpressionSync(
...     from_="work_orders",
...     filter=Equals(
...         ["edge", "type"], {"space": "mySpace", "externalId": "WorkOrder.pump"}
...     ),
...     sync_mode="two_phase",
... )

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.data_modeling.query.Intersection( intersection: 'Sequence[str | SetOperation]', except_: 'SequenceNotStr[str] | None' = None, limit: 'int | None' = None, )

Bases: SetOperation

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.data_modeling.query.NodeOrEdgeResultSetExpression( from_: 'str | None' = None, filter: 'Filter | None' = None, limit: 'int | None' = None, sort: 'list[InstanceSort]' = <factory>, direction: "Literal['outwards', 'inwards']" = 'outwards', chain_to: "Literal['destination', 'source']" = 'destination', ): Bases: ResultSetExpression, ABC

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] | PropertyId | None = None, direction: Literal['outwards', 'inwards'] = 'outwards', chain_to: Literal['destination', 'source'] = 'destination', )

Bases: NodeOrEdgeResultSetExpression

Describes how to query for nodes in the data model.

Parameters:

from (str | None) – Chain your result-expression based on this view.
filter (Filter | None) – Filter the result set based on this filter.
sort (list[InstanceSort] | None) – Sort the result set based on this list of sort criteria.
limit (int | None) – Limit the result set to this number of instances.
through (list[str] | tuple[str, str, str] | PropertyId | None) – Chain your result-expression through this container or view. The property must be a reference to a direct relation property. from_ must be defined. The tuple must be on the form (space, container, property) or (space, view/version, property).
direction (Literal['outwards', 'inwards']) – The direction to use when traversing direct relations. Only applicable when through is specified.
chain_to (Literal['destination', 'source']) – Control which side of the edge to chain to. The chain_to option is only applicable if the result rexpression referenced in from contains edges. source will chain to start if you’re following edges outwards i.e direction=outwards. If you’re following edges inwards i.e direction=inwards, it will chain to end. destination (default) will chain to end if you’re following edges outwards i.e direction=outwards. If you’re following edges inwards i.e, direction=inwards, it will chain to start.

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.data_modeling.query.NodeResultSetExpressionSync( from_: str | None = None, filter: Filter | None = None, limit: int | None = None, through: list[str] | tuple[str, str, str] | PropertyId | None = None, direction: Literal['outwards', 'inwards'] = 'outwards', chain_to: Literal['destination', 'source'] = 'destination', skip_already_deleted: bool = True, sync_mode: Literal['one_phase', 'two_phase', 'no_backfill'] = 'two_phase', backfill_sort: list[InstanceSort] | None = None, )

Bases: ResultSetExpressionSync

Describes how to query for nodes in the data model.

Parameters:

from (str | None) – Chain your result-expression based on this view.
filter (Filter | None) – Filter the result set based on this filter.
limit (int | None) – Limit the result set to this number of instances.
through (list[str] | tuple[str, str, str] | PropertyId | None) – Chain your result-expression through this container or view. The property must be a reference to a direct relation property. from_ must be defined. The tuple must be on the form (space, container, property) or (space, view/version, property).
direction (Literal['outwards', 'inwards']) – The direction to use when traversing direct relations. Only applicable when through is specified.
chain_to (Literal['destination', 'source']) – Control which side of the edge to chain to. The chain_to option is only applicable if the result rexpression referenced in from contains edges. source will chain to start if you’re following edges outwards i.e direction=outwards. If you’re following edges inwards i.e direction=inwards, it will chain to end. destination (default) will chain to end if you’re following edges outwards i.e direction=outwards. If you’re following edges inwards i.e, direction=inwards, it will chain to start.
skip_already_deleted (bool) – If set to False, the API will return instances that have been soft deleted before sync was initiated. Soft deletes that happen after the sync is initiated and a cursor generated, are always included in the result. Soft deleted instances are identified by having deletedTime set.
sync_mode (Literal['one_phase', 'two_phase', 'no_backfill']) – Specify whether to sync instances in a single phase; in a backfill phase followed by live updates, or without any backfill. Defaults to “two_phase”, which is recommended for queries with custom filters. Note that the backfill phase’s filters and sort must be backed by a cursorable index. Use “one_phase” when fetching all instances (or all in specific spaces) for better throughput.
backfill_sort (list[InstanceSort] | None) – Sort the result set during the backfill phase of a two phase sync. Only valid with sync_mode = “two_phase”. The sort must be backed by a cursorable index.

Examples

Import and instantiate:

>>> from cognite.client.data_classes.data_modeling import (
...     NodeResultSetExpressionSync,
... )
>>> from cognite.client.data_classes.filters import SpaceFilter
>>> expr = NodeResultSetExpressionSync(
...     filter=SpaceFilter("mySpace"),
...     sync_mode="two_phase",
... )

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.data_modeling.query.Query(with_: ~collections.abc.MutableMapping[str, ~cognite.client.data_classes.data_modeling.query._T_ResultSetExpression], select: ~collections.abc.MutableMapping[str, ~cognite.client.data_classes.data_modeling.query._T_Select], parameters: ~collections.abc.Mapping[str, str | int | float | bool | dict | list[str] | list[int] | list[float] | list[bool] | list[dict]] = <factory>, cursors: ~collections.abc.Mapping[str, str | None] = <factory>)

Bases: QueryBase[ResultSetExpression, Select]

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

Parameters:

with (MutableMapping[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 (MutableMapping[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 (Mapping[str, PropertyValue]) – 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]) – A dictionary of cursors to use in the query. These allow for pagination.

class cognite.client.data_classes.data_modeling.query.QueryBase(with_: ~collections.abc.MutableMapping[str, ~cognite.client.data_classes.data_modeling.query._T_ResultSetExpression], select: ~collections.abc.MutableMapping[str, ~cognite.client.data_classes.data_modeling.query._T_Select], parameters: ~collections.abc.Mapping[str, str | int | float | bool | dict | list[str] | list[int] | list[float] | list[bool] | list[dict]] = <factory>, cursors: ~collections.abc.Mapping[str, str | None] = <factory>)

Bases: CogniteResource, ABC, Generic[_T_ResultSetExpression, _T_Select]

Abstract base class for normal queries and sync queries

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.data_modeling.query.QueryResult(*args: Any, **kwargs: Any): Bases: UserDict

class cognite.client.data_classes.data_modeling.query.QuerySync(with_: ~collections.abc.MutableMapping[str, ~cognite.client.data_classes.data_modeling.query._T_ResultSetExpression], select: ~collections.abc.MutableMapping[str, ~cognite.client.data_classes.data_modeling.query._T_Select], parameters: ~collections.abc.Mapping[str, str | int | float | bool | dict | list[str] | list[int] | list[float] | list[bool] | list[dict]] = <factory>, cursors: ~collections.abc.Mapping[str, str | None] = <factory>, allow_expired_cursors_and_accept_missed_deletes: bool = False)

Bases: QueryBase[ResultSetExpressionSync, SelectSync]

Sync allows you to do subscribe to changes in instances.

Parameters:

with (MutableMapping[str, ResultSetExpressionSync]) – 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 (MutableMapping[str, SelectSync]) – 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 (Mapping[str, PropertyValue]) – 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]) – A dictionary of cursors to use in the query. These allow for pagination.
allow_expired_cursors_and_accept_missed_deletes (bool) – Sync cursors expire after 3 days because soft-deleted instances are cleaned up after this grace period, so a client using a cursor older than that risks missing deletes. If set to True, the API will allow the use of expired cursors.

Examples

Import and instantiate:

>>> from cognite.client.data_classes.data_modeling import (
...     QuerySync,
...     NodeResultSetExpressionSync,
...     SelectSync,
...     SourceSelector,
...     ViewId,
... )
>>> from cognite.client.data_classes.filters import SpaceFilter
>>> view_id = ViewId("mySpace", "myView", "v1")
>>> sync_query = QuerySync(
...     with_={"assets": NodeResultSetExpressionSync(filter=SpaceFilter("mySpace"))},
...     select={"assets": SelectSync([SourceSelector(view_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.data_modeling.query.ResultSetExpression: Bases: ResultSetExpressionBase, ABC

class cognite.client.data_classes.data_modeling.query.ResultSetExpressionBase: Bases: CogniteResource, ABC

class cognite.client.data_classes.data_modeling.query.ResultSetExpressionSync( from_: 'str | None' = None, filter: 'Filter | None' = None, limit: 'int | None' = None, direction: "Literal['outwards', 'inwards']" = 'outwards', chain_to: "Literal['destination', 'source']" = 'destination', skip_already_deleted: 'bool' = True, sync_mode: 'SyncMode' = 'two_phase', backfill_sort: 'list[InstanceSort]' = <factory>, ): Bases: ResultSetExpressionBase, ABC

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

Bases: SelectBase

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.data_modeling.query.SelectBase(sources: 'list[SourceSelector]' = <factory>)

Bases: CogniteResource, ABC

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.data_modeling.query.SelectSync( sources: list[~cognite.client.data_classes.data_modeling.query.SourceSelector] = <factory>, )

Bases: SelectBase

Select expression for a sync query, specifying which view properties to include.

The properties list on each SourceSelector controls which view properties are returned for matching instances:

["*"] — wildcard; returns all properties defined in the view. Convenient but
transfers more data per instance.
An explicit list such as ["name", "description"] — returns only those named
properties. Reduces payload size when only a subset is needed.

Examples

All properties from a view:

>>> from cognite.client.data_classes.data_modeling import (
...     SelectSync,
...     SourceSelector,
...     ViewId,
... )
>>> view_id = ViewId("mySpace", "myView", "v1")
>>> select_all = SelectSync([SourceSelector(view_id, ["*"])])

Only specific properties:

>>> select_some = SelectSync([SourceSelector(view_id, ["name", "description"])])

class cognite.client.data_classes.data_modeling.query.SetOperation: Bases: ResultSetExpression, ABC

class cognite.client.data_classes.data_modeling.query.SourceSelector( source: 'ViewId', properties: 'list[str] | None' = None, target_units: 'list[TargetUnit] | None' = None, )

Bases: CogniteResource

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.data_modeling.query.Union( union: 'Sequence[str | SetOperation]', except_: 'SequenceNotStr[str] | None' = None, limit: 'int | None' = None, )

Bases: SetOperation

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.data_modeling.query.UnionAll( union_all: 'Sequence[str | SetOperation]', except_: 'SequenceNotStr[str] | None' = None, limit: 'int | None' = None, )

Bases: SetOperation

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]

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(space: 'str', external_id: 'str'): Bases: DataModelingId

class cognite.client.data_classes.data_modeling.ids.DataModelId(space: 'str', external_id: 'str', version: 'str | None' = None): Bases: VersionedDataModelingId

class cognite.client.data_classes.data_modeling.ids.DataModelingId(space: 'str', external_id: 'str'): 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.NodeId(space: 'str', external_id: 'str'): Bases: InstanceId

class cognite.client.data_classes.data_modeling.ids.PropertyId(source: 'ViewId | ContainerId', property: 'str')

Bases: CogniteResource

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.data_modeling.ids.VersionedDataModelingId( space: 'str', external_id: 'str', version: 'str | None' = None, ): 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(space: 'str', external_id: 'str', version: 'str | None' = None): Bases: VersionedDataModelingId