Create asset hierarchy

async AsyncCogniteClient.assets.create_hierarchy(
assets: Sequence[AssetWrite] | AssetHierarchy,
*,
upsert: bool = False,
upsert_mode: Literal['patch', 'replace'] = 'patch',
) → AssetList

Create an asset hierarchy with validation.

This helper function makes it easy to insert large asset hierarchies. It solves the problem of topological insertion order, i.e. a parent asset must exist before it can be referenced by any ‘children’ assets. You may pass any number of partial- or full hierarchies: there are no requirements on the number of root assets, so you may pass zero, one or many (same goes for the non-root assets).

Parameters:

  • assets (Sequence[AssetWrite] | AssetHierarchy) – List of assets to create or an instance of AssetHierarchy.

  • upsert (bool) – If used, already existing assets will be updated instead of an exception being raised. You may control how updates are applied with the ‘upsert_mode’ argument.

  • upsert_mode (Literal['patch', 'replace']) – Only applicable with upsert. Pass ‘patch’ to only update fields with non-null values (default), or ‘replace’ to do full updates (unset fields become null or empty).

Returns:

Created (and possibly updated) asset hierarchy

Return type:

AssetList

Prior to insertion, this function will run validation on the given assets and raise an error if any of the following issues are found:

  1. Any assets are invalid (category: invalid):

    • Missing external ID.

    • Missing a valid name.

    • Has an ID set (note: you may not pass Asset, use AssetWrite)

  2. Any asset duplicates exist (category: duplicates)

  3. Any assets have an ambiguous parent link (category: unsure_parents)

  4. Any group of assets form a cycle, e.g. A->B->A (category: cycles)

As part of validation there is a fifth category that is ignored when using this method (for backwards compatibility) and that is orphan assets. These are assets linking a parent by an identifier that is not present among the given assets, and as such, might contain links we are unable to vet ahead of insertion. These are thus assumed to be valid, but may fail.

Tip

The different categories specified above corresponds to the name of the attribute you might access on the raised error to get the collection of ‘bad’ assets falling in that group, e.g. error.duplicates.

Note

Updating external_id via upsert is not supported (and will not be supported). Use AssetsAPI.update instead.

Warning

The API does not natively support upsert, so the SDK has to simulate the behaviour at the cost of some insertion speed.

Be careful when moving assets to new parents via upsert: Please do so only by specifying parent_external_id (instead of parent_id) to avoid race conditions in insertion order (temporary cycles might form since we can only make changes to 1000 assets at the time).

Examples

Create an asset hierarchy:

>>> from cognite.client import CogniteClient
>>> from cognite.client.data_classes import AssetWrite
>>> client = CogniteClient()
>>> # async_client = AsyncCogniteClient()  # another option
>>> assets = [
...     AssetWrite(external_id="root", name="root"),
...     AssetWrite(external_id="child1", parent_external_id="root", name="child1"),
...     AssetWrite(external_id="child2", parent_external_id="root", name="child2"),
... ]
>>> res = client.assets.create_hierarchy(assets)

Create an asset hierarchy, but run update for existing assets:

>>> res = client.assets.create_hierarchy(assets, upsert=True, upsert_mode="patch")

Patch will only update the parameters you have defined on your assets. Note that specifically setting something to None is the same as not setting it. For metadata, this will extend your existing data, only overwriting when keys overlap. For labels the behaviour is mostly the same, existing are left untouched, and your new ones are simply added.

You may also pass upsert_mode="replace" to make sure the updated assets look identical to the ones you passed to the method. For both metadata and labels this will clear out all existing, before (potentially) adding the new ones.

If the hierarchy validation for some reason fail, you may inspect all the issues that were found by catching CogniteAssetHierarchyError:

>>> from cognite.client.exceptions import CogniteAssetHierarchyError
>>> try:
...     res = client.assets.create_hierarchy(assets)
... except CogniteAssetHierarchyError as err:
...     if err.invalid:
...         ...  # do something

In addition to invalid, you may inspect duplicates, unsure_parents, orphans and cycles. Note that cycles are not available if any of the other basic issues exist, as the search for cyclical references requires a clean asset hierarchy to begin with.

You may also wrap the create_hierarchy() call in a try-except to get information if any of the assets fails to be created (assuming a valid hierarchy):

>>> from cognite.client.exceptions import CogniteAPIError
>>> try:
...     client.assets.create_hierarchy(assets)
... except CogniteAPIError as err:
...     created = err.successful
...     maybe_created = err.unknown
...     not_created = err.failed

Here’s a slightly longer explanation of the different groups:

  • err.successful: Which assets were created (request yielded a 201)

  • err.unknown: Which assets may have been created (request yielded 5xx)

  • err.failed: Which assets were not created (request yielded 4xx, or was a descendant of an asset with unknown status)

The preferred way to create an asset hierarchy, is to run validation prior to insertion. You may do this by using the AssetHierarchy class. It will by default consider orphan assets to be problematic (but accepts the boolean parameter ignore_orphans), contrary to how create_hierarchy works (which accepts them in order to be backwards-compatible). It also provides helpful methods to create reports of any issues found, check out validate_and_report:

>>> from cognite.client.data_classes import AssetHierarchy
>>> from pathlib import Path
>>> hierarchy = AssetHierarchy(assets)
>>> if hierarchy.is_valid():
...     res = client.assets.create_hierarchy(hierarchy)
... else:
...     hierarchy.validate_and_report(output_file=Path("report.txt"))