Apply instances

async AsyncCogniteClient.data_modeling.instances.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,
... )
>>> client = CogniteClient()
>>> # async_client = AsyncCogniteClient()  # another option
>>> node = NodeApply("mySpace", "myNodeId")
>>> res = client.data_modeling.instances.apply(node)

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

>>> from cognite.client.data_classes.data_modeling import (
...     ContainerId,
...     EdgeApply,
...     NodeOrEdgeData,
...     NodeApply,
...     ViewId,
... )
>>> work_order = NodeApply(
...     space="industrial",
...     external_id="work_order:123",
...     sources=[
...         # Insert data through a view
...         NodeOrEdgeData(
...             ViewId("mySpace", "WorkOrderView", "v1"),
...             {"title": "Repair pump", "createdYear": 2023},
...         )
...     ],
... )
>>> pump = NodeApply(
...     space="industrial",
...     external_id="pump:456",
...     sources=[
...         # Insert data directly to the container
...         NodeOrEdgeData(
...             ContainerId("mySpace", "PumpContainer"),
...             {"name": "Pump 456", "location": "Subsea"},
...         )
...     ],
... )
... # This is one-to-many edge, in this case from a work order to a pump
>>> work_order_to_pump = EdgeApply(
...     space="industrial",
...     external_id="relation:work_order:123:pump:456",
...     type=("industrial", "relates-to"),
...     start_node=("industrial", "work_order:123"),
...     end_node=("industrial", "pump:456"),
... )
>>> res = client.data_modeling.instances.apply([work_order, pump], [work_order_to_pump])

Create new edge and automatically create end nodes.

>>> from cognite.client.data_classes.data_modeling import EdgeApply
>>> work_order_to_pump = EdgeApply(
...     space="industrial",
...     external_id="relation:work_order:123:pump:456",
...     type=("industrial", "relates-to"),
...     start_node=("industrial", "work_order:123"),
...     end_node=("industrial", "pump:456"),
... )
>>> res = client.data_modeling.instances.apply(
...     edges=work_order_to_pump,
...     auto_create_start_nodes=True,
...     auto_create_end_nodes=True,
... )

Using helper function to create valid graphql timestamp for a datetime object:

>>> from cognite.client.utils import datetime_to_ms_iso_timestamp
>>> from datetime import datetime, timezone
>>> my_date = datetime(2020, 3, 14, 15, 9, 26, 535000, tzinfo=timezone.utc)
>>> data_model_timestamp = datetime_to_ms_iso_timestamp(
...     my_date
... )  # "2020-03-14T15:09:26.535+00:00"

Create a typed node apply. Any property that you want to look up by a different attribute name, e.g. you want my_node.birth_year to return the data for property birthYear, must use the PropertyOptions as shown below. We strongly suggest you use snake_cased attribute names, as is done here:

>>> from cognite.client.data_classes.data_modeling import TypedNodeApply, PropertyOptions
>>> class PersonApply(TypedNodeApply):
...     birth_year = PropertyOptions(identifier="birthYear")
...
...     def __init__(self, space: str, external_id, name: str, birth_year: int):
...         super().__init__(space, external_id, type=("sp_model_space", "Person"))
...         self.name = name
...         self.birth_year = birth_year
...
...     def get_source(self):
...         return ViewId("sp_model_space", "Person", "v1")
>>> person = PersonApply("sp_date_space", "my_person", "John Doe", 1980)
>>> res = client.data_modeling.instances.apply(nodes=person)