Sync instances

async AsyncCogniteClient.data_modeling.instances.sync( query: QuerySync, include_typing: bool = False, debug: DebugParameters | None = None, ) → QueryResult

Subscription to changes for nodes/edges.

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

Parameters:

query (QuerySync) – Query.
include_typing (bool) – Should we return property type information as part of the result?
debug (DebugParameters | None) – Debug settings for profiling and troubleshooting.

Returns:

The resulting nodes and/or edges from the query.

Return type:

QueryResult

Examples

Query all pumps connected to work orders created before 2023, sorted by 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
>>> client = CogniteClient()
>>> # async_client = AsyncCogniteClient()  # another option
>>> work_order_id = ViewId("mySpace", "WorkOrderView", "v1")
>>> pump_id = ViewId("mySpace", "PumpView", "v1")
>>> query = Query(
...     with_={
...         "work_orders": NodeResultSetExpression(
...             filter=Range(work_order_id.as_property_ref("createdYear"), lt=2023)
...         ),
...         "work_orders_to_pumps": EdgeResultSetExpression(
...             from_="work_orders",
...             filter=Equals(
...                 ["edge", "type"],
...                 {"space": work_order_id.space, "externalId": "WorkOrder.asset"},
...             ),
...         ),
...         "pumps": NodeResultSetExpression(from_="work_orders_to_pumps"),
...     },
...     select={
...         "pumps": Select(
...             [SourceSelector(pump_id, ["name"])],
...             sort=[InstanceSort(pump_id.as_property_ref("name"))],
...         ),
...     },
... )
>>> res = client.data_modeling.instances.sync(query)
>>> # Added a new work order with pumps created before 2023
>>> query.cursors = res.cursors
>>> res_new = client.data_modeling.instances.sync(query)

In the last example, the res_new will only contain the pumps that have been added with the new work order.

To debug and/or profile your query, you can use the debug parameter:

>>> from cognite.client.data_classes.data_modeling.debug import DebugParameters
>>> debug_params = DebugParameters(
...     emit_results=False,
...     include_plan=True,  # Include the postgres execution plan
...     include_translated_query=True,  # Include the internal representation of the query.
...     profile=True,
... )
>>> res = client.data_modeling.instances.sync(query, debug=debug_params)
>>> print(res.debug)