Assets

Retrieve an asset by id

AssetsAPI.retrieve(id: int | None = None, external_id: str | None = None) → Asset | None

Retrieve a single asset by id.

Parameters

id (int | None) – ID
external_id (str | None) – External ID

Returns

Requested asset or None if it does not exist.

Return type

Asset | None

Examples

Get asset by id:

>>> from cognite.client import CogniteClient
>>> client = CogniteClient()
>>> res = client.assets.retrieve(id=1)

Get asset by external id:

>>> from cognite.client import CogniteClient
>>> client = CogniteClient()
>>> res = client.assets.retrieve(external_id="1")

Retrieve multiple assets by id

AssetsAPI.retrieve_multiple(ids: Sequence[int] | None = None, external_ids: SequenceNotStr[str] | None = None, ignore_unknown_ids: bool = False) → AssetList

Retrieve multiple assets by id.

Parameters

ids (Sequence[int] | None) – IDs
external_ids (SequenceNotStr[str] | None) – External IDs
ignore_unknown_ids (bool) – Ignore IDs and external IDs that are not found rather than throw an exception.

Returns

The requested assets.

Return type

AssetList

Examples

Get assets by id:

>>> from cognite.client import CogniteClient
>>> client = CogniteClient()
>>> res = client.assets.retrieve_multiple(ids=[1, 2, 3])

Get assets by external id:

>>> from cognite.client import CogniteClient
>>> client = CogniteClient()
>>> res = client.assets.retrieve_multiple(external_ids=["abc", "def"], ignore_unknown_ids=True)

Retrieve an asset subtree

AssetsAPI.retrieve_subtree(id: int | None = None, external_id: str | None = None, depth: int | None = None) → AssetList

Retrieve the subtree for this asset up to a specified depth.

Parameters

id (int | None) – Id of the root asset in the subtree.
external_id (str | None) – External id of the root asset in the subtree.
depth (int | None) – Retrieve assets up to this depth below the root asset in the subtree. Omit to get the entire subtree.

Returns

The requested assets or empty AssetList if asset does not exist.

Return type

AssetList

List assets

List assets

Parameters

name (str | None) – Name of asset. Often referred to as tag.
parent_ids (Sequence[int] | None) – Return only the direct descendants of the specified assets.
parent_external_ids (SequenceNotStr[str] | None) – Return only the direct descendants of the specified assets.
asset_subtree_ids (int | Sequence[int] | None) – Only include assets in subtrees rooted at any of the specified assetIds. If the total size of the given subtrees exceeds 100,000 assets, an error will be returned.
asset_subtree_external_ids (str | SequenceNotStr[str] | None) – Only include assets in subtrees rooted at any of the specified assetExternalIds. If the total size of the given subtrees exceeds 100,000 assets, an error will be returned.
data_set_ids (int | Sequence[int] | None) – Return only assets in the specified data set(s) with this id / these ids.
data_set_external_ids (str | SequenceNotStr[str] | None) – Return only assets in the specified data set(s) with this external id / these external ids.
labels (LabelFilter | None) – Return only the assets matching the specified label filter.
geo_location (GeoLocationFilter | None) – Only include files matching the specified geographic relation.
metadata (dict[str, str] | None) – Custom, application specific metadata. String key -> String value.
source (str | None) – The source of this asset.
created_time (dict[str, Any] | TimestampRange | None) – Range between two timestamps. Possible keys are min and max, with values given as time stamps in ms.
last_updated_time (dict[str, Any] | TimestampRange | None) – Range between two timestamps. Possible keys are min and max, with values given as time stamps in ms.
root (bool | None) – filtered assets are root assets or not.
external_id_prefix (str | None) – Filter by this (case-sensitive) prefix for the external ID.
aggregated_properties (Sequence[AggregateAssetProperty] | None) – Set of aggregated properties to include. Options are childCount, path, depth.
partitions (int | None) – Retrieve resources in parallel using this number of workers (values up to 10 allowed), limit must be set to None (or -1).
limit (int | None) – Maximum number of assets to return. Defaults to 25. Set to -1, float(“inf”) or None to return all items.
advanced_filter (Filter | dict[str, Any] | None) – Advanced filter query using the filter DSL (Domain Specific Language). It allows defining complex filtering expressions that combine simple operations, such as equals, prefix, exists, etc., using boolean operators and, or, and not. See examples below for usage.
sort (SortSpec | list[SortSpec] | None) – The criteria to sort by. Defaults to desc for _score_ and asc for all other properties. Sort is not allowed if partitions is used.

Returns

List of requested assets

Return type

AssetList

Note

When using partitions, there are few considerations to keep in mind:

limit has to be set to None (or -1).
API may reject requests if you specify more than 10 partitions. When Cognite enforces this behavior, the requests result in a 400 Bad Request status.
Partitions are done independently of sorting: there’s no guarantee of the sort order between elements from different partitions. For this reason providing a sort parameter when using partitions is not allowed.

Examples

List assets:

>>> from cognite.client import CogniteClient
>>> client = CogniteClient()
>>> asset_list = client.assets.list(limit=5)

Iterate over assets:

>>> from cognite.client import CogniteClient
>>> client = CogniteClient()
>>> for asset in client.assets:
...     asset # do something with the asset

Iterate over chunks of assets to reduce memory load:

>>> from cognite.client import CogniteClient
>>> client = CogniteClient()
>>> for asset_list in client.assets(chunk_size=2500):
...     asset_list # do something with the assets

Filter assets based on labels:

>>> from cognite.client import CogniteClient
>>> from cognite.client.data_classes import LabelFilter
>>> client = CogniteClient()
>>> my_label_filter = LabelFilter(contains_all=["PUMP", "VERIFIED"])
>>> asset_list = client.assets.list(labels=my_label_filter)

Using advanced filter, find all assets that have a metadata key ‘timezone’ starting with ‘Europe’, and sort by external id ascending:

>>> from cognite.client import CogniteClient
>>> from cognite.client.data_classes import filters
>>> client = CogniteClient()
>>> in_timezone = filters.Prefix(["metadata", "timezone"], "Europe")
>>> res = client.assets.list(advanced_filter=in_timezone, sort=("external_id", "asc"))

Note that you can check the API documentation above to see which properties you can filter on with which filters.

To make it easier to avoid spelling mistakes and easier to look up available properties for filtering and sorting, you can also use the AssetProperty and SortableAssetProperty Enums.

>>> from cognite.client import CogniteClient
>>> from cognite.client.data_classes import filters
>>> from cognite.client.data_classes.assets import AssetProperty, SortableAssetProperty
>>> client = CogniteClient()
>>> in_timezone = filters.Prefix(AssetProperty.metadata_key("timezone"), "Europe")
>>> res = client.assets.list(
...     advanced_filter=in_timezone,
...     sort=(SortableAssetProperty.external_id, "asc"))

Combine filter and advanced filter:

>>> from cognite.client import CogniteClient
>>> from cognite.client.data_classes import filters
>>> client = CogniteClient()
>>> not_instrument_lvl5 = filters.And(
...    filters.ContainsAny("labels", ["Level5"]),
...    filters.Not(filters.ContainsAny("labels", ["Instrument"]))
... )
>>> res = client.assets.list(asset_subtree_ids=[123456], advanced_filter=not_instrument_lvl5)

Aggregate assets

AssetsAPI.aggregate(filter: AssetFilter | dict[str, Any] | None = None) → list[CountAggregate]

Aggregate assets

Parameters: filter (AssetFilter | dict[str, Any] | None) – Filter on assets with strict matching.
Returns: List of asset aggregates
Return type: list[CountAggregate]

Examples

Aggregate assets:

>>> from cognite.client import CogniteClient
>>> client = CogniteClient()
>>> aggregate_by_prefix = client.assets.aggregate(filter={"external_id_prefix": "prefix"})

Aggregate Asset Count

Count of assets matching the specified filters.

Parameters

property (AssetPropertyLike | None) – If specified, get an approximate number of asset with a specific property (property is not null) and matching the filters.
advanced_filter (Filter | dict[str, Any] | None) – The advanced filter to narrow down the assets to count.
filter (AssetFilter | dict[str, Any] | None) – The filter to narrow down the assets to count (strict matching).

Returns

The number of assets matching the specified filters.

Return type

int

Examples:

Count the number of assets in your CDF project:

>>> from cognite.client import CogniteClient
>>> client = CogniteClient()
>>> count = client.assets.aggregate_count()

Count the number of assets with the metadata key “timezone” in your CDF project:

>>> from cognite.client import CogniteClient
>>> from cognite.client.data_classes.filters import ContainsAny
>>> from cognite.client.data_classes.assets import AssetProperty
>>> client = CogniteClient()
>>> has_timezone = ContainsAny(AssetProperty.metadata, "timezone")
>>> asset_count = client.assets.aggregate_count(advanced_filter=has_timezone)

Aggregate Asset Value Cardinality

Find approximate property count for assets.

Parameters

property (AssetPropertyLike) – The property to count the cardinality of.
advanced_filter (Filter | dict[str, Any] | None) – The advanced filter to narrow down assets.
aggregate_filter (AggregationFilter | dict[str, Any] | None) – The filter to apply to the resulting buckets.
filter (AssetFilter | dict[str, Any] | None) – The filter to narrow down assets (strict matching).

Returns

The number of properties matching the specified filters and search.

Return type

int

Examples

Count the number of labels used by assets in your CDF project:

>>> from cognite.client import CogniteClient
>>> from cognite.client.data_classes.assets import AssetProperty
>>> client = CogniteClient()
>>> label_count = client.assets.aggregate_cardinality_values(AssetProperty.labels)

Count the number of timezones (metadata key) for assets with the word “critical” in the description in your CDF project:

>>> from cognite.client import CogniteClient
>>> from cognite.client.data_classes.filters import Search
>>> from cognite.client.data_classes.assets import AssetProperty
>>> client = CogniteClient()
>>> is_critical = Search(AssetProperty.description, "critical")
>>> critical_assets = client.assets.aggregate_cardinality_values(
...     AssetProperty.metadata_key("timezone"),
...     advanced_filter=is_critical)

Aggregate Asset Property Cardinality

Find approximate paths count for assets.

Parameters

path (AssetPropertyLike) – The scope in every document to aggregate properties. The only value allowed now is [“metadata”]. It means to aggregate only metadata properties (aka keys).
advanced_filter (Filter | dict[str, Any] | None) – The advanced filter to narrow down assets.
aggregate_filter (AggregationFilter | dict[str, Any] | None) – The filter to apply to the resulting buckets.
filter (AssetFilter | dict[str, Any] | None) – The filter to narrow down assets (strict matching).

Returns

The number of properties matching the specified filters.

Return type

int

Examples

Count the number of unique metadata keys used by assets in your CDF project:

>>> from cognite.client import CogniteClient
>>> from cognite.client.data_classes.assets import AssetProperty
>>> client = CogniteClient()
>>> key_count = client.assets.aggregate_cardinality_properties(AssetProperty.metadata)

Aggregate Asset Unique Values

Get unique properties with counts for assets.

Note

In the case of text fields, the values are aggregated in a case-insensitive manner.

Parameters

property (AssetPropertyLike) – The property to group by.
advanced_filter (Filter | dict[str, Any] | None) – The advanced filter to narrow down assets.
aggregate_filter (AggregationFilter | dict[str, Any] | None) – The filter to apply to the resulting buckets.
filter (AssetFilter | dict[str, Any] | None) – The filter to narrow down assets (strict matching).

Returns

List of unique values of assets matching the specified filters and search.

Return type

UniqueResultList

Examples:

Get the timezones (metadata key) with count for your assets in your CDF project:

>>> from cognite.client import CogniteClient
>>> from cognite.client.data_classes.assets import AssetProperty
>>> client = CogniteClient()
>>> result = client.assets.aggregate_unique_values(AssetProperty.metadata_key("timezone"))
>>> print(result.unique)

Get the different labels with count used for assets created after 2020-01-01 in your CDF project:

>>> from cognite.client import CogniteClient
>>> from cognite.client.data_classes import filters as flt
>>> from cognite.client.data_classes.assets import AssetProperty
>>> from cognite.client.utils import timestamp_to_ms
>>> from datetime import datetime
>>> client = CogniteClient()
>>> created_after_2020 = flt.Range(AssetProperty.created_time, gte=timestamp_to_ms(datetime(2020, 1, 1)))
>>> result = client.assets.aggregate_unique_values(AssetProperty.labels, advanced_filter=created_after_2020)
>>> print(result.unique)

Get the different labels with count for assets updated after 2020-01-01 in your CDF project, but exclude all labels that starts with “test”:

>>> from cognite.client import CogniteClient
>>> from cognite.client.data_classes.assets import AssetProperty
>>> from cognite.client.data_classes import aggregations as aggs
>>> from cognite.client.data_classes import filters as flt
>>> client = CogniteClient()
>>> not_test = aggs.Not(aggs.Prefix("test"))
>>> created_after_2020 = flt.Range(AssetProperty.last_updated_time, gte=timestamp_to_ms(datetime(2020, 1, 1)))
>>> result = client.assets.aggregate_unique_values(AssetProperty.labels, advanced_filter=created_after_2020, aggregate_filter=not_test)
>>> print(result.unique)

Aggregate Asset Unique Properties

Get unique paths with counts for assets.

Note

In the case of text fields, the values are aggregated in a case-insensitive manner.

Parameters

path (AssetPropertyLike) – The scope in every document to aggregate properties. The only value allowed now is [“metadata”]. It means to aggregate only metadata properties (aka keys).
advanced_filter (Filter | dict[str, Any] | None) – The advanced filter to narrow down assets.
aggregate_filter (AggregationFilter | dict[str, Any] | None) – The filter to apply to the resulting buckets.
filter (AssetFilter | dict[str, Any] | None) – The filter to narrow down assets (strict matching).

Returns

List of unique values of assets matching the specified filters and search.

Return type

UniqueResultList

Examples

Get the metadata keys with counts for your assets in your CDF project:

>>> from cognite.client import CogniteClient
>>> from cognite.client.data_classes.assets import AssetProperty
>>> client = CogniteClient()
>>> result = client.assets.aggregate_unique_properties(AssetProperty.metadata)

Search for assets

Search for assets Primarily meant for human-centric use-cases and data exploration, not for programs, since matching and ordering may change over time. Use the list function if stable or exact matches are required.

Parameters

name (str | None) – Fuzzy match on name.
description (str | None) – Fuzzy match on description.
query (str | None) – Whitespace-separated terms to search for in assets. Does a best-effort fuzzy search in relevant fields (currently name and description) for variations of any of the search terms, and orders results by relevance.
filter (AssetFilter | dict[str, Any] | None) – Filter to apply. Performs exact match on these fields.
limit (int) – Maximum number of results to return.

Returns

List of requested assets

Return type

AssetList

Examples

Search for assets by fuzzy search on name:

>>> from cognite.client import CogniteClient
>>> client = CogniteClient()
>>> res = client.assets.search(name="some name")

Search for assets by exact search on name:

>>> from cognite.client import CogniteClient
>>> client = CogniteClient()
>>> res = client.assets.search(filter={"name": "some name"})

Search for assets by improved multi-field fuzzy search:

>>> from cognite.client import CogniteClient
>>> client = CogniteClient()
>>> res = client.assets.search(query="TAG 30 XV")

Search for assets using multiple filters, finding all assets with name similar to xyz with parent asset 123 or 456 with source some source:

>>> from cognite.client import CogniteClient
>>> client = CogniteClient()
>>> res = client.assets.search(name="xyz",filter={"parent_ids": [123,456],"source": "some source"})

Search for an asset with an attached label:

>>> from cognite.client import CogniteClient
>>> client = CogniteClient()
>>> my_label_filter = LabelFilter(contains_all=["PUMP"])
>>> res = client.assets.search(name="xyz",filter=AssetFilter(labels=my_label_filter))

Create assets

AssetsAPI.create(asset: Sequence[Asset] | Sequence[AssetWrite]) → AssetList

AssetsAPI.create(asset: Asset | AssetWrite) → Asset

Create one or more assets.

You can create an arbitrary number of assets, and the SDK will split the request into multiple requests. When specifying parent-child relation between assets using parentExternalId the link will be resvoled into an internal ID and stored as parentId.

Parameters: asset (Asset | AssetWrite | Sequence[Asset] | Sequence[AssetWrite]) – Asset or list of assets to create.
Returns: Created asset(s)
Return type: Asset | AssetList

Examples

Create new assets:

>>> from cognite.client import CogniteClient
>>> from cognite.client.data_classes import AssetWrite
>>> client = CogniteClient()
>>> assets = [AssetWrite(name="asset1"), AssetWrite(name="asset2")]
>>> res = client.assets.create(assets)

Create asset with label:

>>> from cognite.client import CogniteClient
>>> from cognite.client.data_classes import Asset, Label
>>> client = CogniteClient()
>>> asset = AssetWrite(name="my_pump", labels=[Label(external_id="PUMP")])
>>> res = client.assets.create(asset)

Create asset hierarchy

AssetsAPI.create_hierarchy(assets: Sequence[Asset | 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[Asset | 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:

Any assets are invalid (category: invalid):

Missing external ID.

Missing a valid name.

Has an ID set.

Any asset duplicates exist (category: duplicates)

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

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 Asset
>>> client = CogniteClient()
>>> assets = [
...     Asset(external_id="root", name="root"),
...     Asset(external_id="child1", parent_external_id="root", name="child1"),
...     Asset(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"))

Delete assets

AssetsAPI.delete(id: int | Sequence[int] | None = None, external_id: str | SequenceNotStr[str] | None = None, recursive: bool = False, ignore_unknown_ids: bool = False) → None

Delete one or more assets

Parameters

id (int | Sequence[int] | None) – Id or list of ids
external_id (str | SequenceNotStr[str] | None) – External ID or list of external ids
recursive (bool) – Recursively delete whole asset subtrees under given ids. Defaults to False.
ignore_unknown_ids (bool) – Ignore IDs and external IDs that are not found rather than throw an exception.

Examples

Delete assets by id or external id:

>>> from cognite.client import CogniteClient
>>> client = CogniteClient()
>>> client.assets.delete(id=[1,2,3], external_id="3")

Filter assets

Advanced filter assets

Advanced filter lets you create complex filtering expressions that combine simple operations, such as equals, prefix, exists, etc., using boolean operators and, or, and not. It applies to basic fields as well as metadata.

Parameters

filter (Filter | dict) – Filter to apply.
sort (SortSpec | list[SortSpec] | None) – The criteria to sort by. Can be up to two properties to sort by default to ascending order.
aggregated_properties (Sequence[AggregateAssetProperty] | None) – Set of aggregated properties to include. Options are childCount, path, depth.
limit (int | None) – Maximum number of results to return. Defaults to 25. Set to -1, float(“inf”) or None to return all items.

Returns

List of assets that match the filter criteria.

Return type

AssetList

Examples

Find all assets that have a metadata key ‘timezone’ starting with ‘Europe’, and sort by external id ascending:

>>> from cognite.client import CogniteClient
>>> from cognite.client.data_classes import filters
>>> client = CogniteClient()
>>> in_timezone = filters.Prefix(["metadata", "timezone"], "Europe")
>>> res = client.assets.filter(filter=in_timezone, sort=("external_id", "asc"))

Note that you can check the API documentation above to see which properties you can filter on with which filters.

To make it easier to avoid spelling mistakes and easier to look up available properties for filtering and sorting, you can also use the AssetProperty and SortableAssetProperty Enums.

>>> from cognite.client import CogniteClient
>>> from cognite.client.data_classes import filters
>>> from cognite.client.data_classes.assets import AssetProperty, SortableAssetProperty
>>> client = CogniteClient()
>>> in_timezone = filters.Prefix(AssetProperty.metadata_key("timezone"), "Europe")
>>> res = client.assets.filter(
...     filter=in_timezone,
...     sort=(SortableAssetProperty.external_id, "asc"))

Update assets

AssetsAPI.update(item: Sequence[Asset | AssetWrite | AssetUpdate]) → AssetList

AssetsAPI.update(item: Asset | AssetWrite | AssetUpdate) → Asset

Update one or more assets Labels can be added, removed or replaced (set). Note that set operation deletes all the existing labels and adds the new specified labels.

Parameters: item (Asset | AssetWrite | AssetUpdate | Sequence[Asset | AssetWrite | AssetUpdate]) – Asset(s) to update
Returns: Updated asset(s)
Return type: Asset | AssetList

Examples

Perform a partial update on an asset, updating the description and adding a new field to metadata:

>>> from cognite.client import CogniteClient
>>> from cognite.client.data_classes import AssetUpdate
>>> client = CogniteClient()
>>> my_update = AssetUpdate(id=1).description.set("New description").metadata.add({"key": "value"})
>>> res1 = client.assets.update(my_update)
>>> # Remove an already set field like so
>>> another_update = AssetUpdate(id=1).description.set(None)
>>> res2 = client.assets.update(another_update)

Remove the metadata on an asset:

>>> from cognite.client import CogniteClient
>>> from cognite.client.data_classes import AssetUpdate
>>> client = CogniteClient()
>>> my_update = AssetUpdate(id=1).metadata.add({"key": "value"})
>>> res1 = client.assets.update(my_update)
>>> another_update = AssetUpdate(id=1).metadata.set(None)
>>> # The same result can be achieved with:
>>> another_update2 = AssetUpdate(id=1).metadata.set({})
>>> res2 = client.assets.update(another_update)

Attach labels to an asset:

>>> from cognite.client import CogniteClient
>>> from cognite.client.data_classes import AssetUpdate
>>> client = CogniteClient()
>>> my_update = AssetUpdate(id=1).labels.add(["PUMP", "VERIFIED"])
>>> res = client.assets.update(my_update)

Detach a single label from an asset:

>>> from cognite.client import CogniteClient
>>> from cognite.client.data_classes import AssetUpdate
>>> client = CogniteClient()
>>> my_update = AssetUpdate(id=1).labels.remove("PUMP")
>>> res = client.assets.update(my_update)

Replace all labels for an asset:

>>> from cognite.client import CogniteClient
>>> from cognite.client.data_classes import AssetUpdate
>>> client = CogniteClient()
>>> my_update = AssetUpdate(id=1).labels.set("PUMP")
>>> res = client.assets.update(my_update)

Upsert assets

AssetsAPI.upsert(item: Sequence[Asset | AssetWrite], mode: Literal['patch', 'replace'] = 'patch') → AssetList

AssetsAPI.upsert(item: Asset | AssetWrite, mode: Literal['patch', 'replace'] = 'patch') → Asset

Upsert assets, i.e., update if it exists, and create if it does not exist.

Note this is a convenience method that handles the upserting for you by first calling update on all items, and if any of them fail because they do not exist, it will create them instead.

For more details, see Upsert.

Parameters

item (Asset | AssetWrite | Sequence[Asset | AssetWrite]) – Asset or list of assets to upsert.
mode (Literal["patch", "replace"]) – Whether to patch or replace in the case the assets are existing. If you set ‘patch’, the call will only update fields with non-null values (default). Setting ‘replace’ will unset any fields that are not specified.

Returns

The upserted asset(s).

Return type

Asset | AssetList

Examples

Upsert for assets:

>>> from cognite.client import CogniteClient
>>> from cognite.client.data_classes import Asset
>>> client = CogniteClient()
>>> existing_asset = client.assets.retrieve(id=1)
>>> existing_asset.description = "New description"
>>> new_asset = Asset(external_id="new_asset", description="New asset")
>>> res = client.assets.upsert([existing_asset, new_asset], mode="replace")

Asset Data classes

class cognite.client.data_classes.assets.AggregateResultItem(child_count: int | None = None, depth: int | None = None, path: list[dict[str, Any]] | None = None, **_: Any)

Bases: CogniteObject

Aggregated metrics of the asset

Parameters

child_count (int | None) – Number of direct descendants for the asset
depth (int | None) – Asset path depth (number of levels below root node).
path (list[dict[str, Any]] | None) – IDs of assets on the path to the asset.
**_ (Any) – No description.

Bases: AssetCore

A representation of a physical asset, for example, a factory or a piece of equipment. This is the read version of the Asset class, it is used when retrieving assets from the Cognite API.

Parameters

external_id (str | None) – The external ID provided by the client. Must be unique for the resource type.
name (str | None) – The name of the asset.
parent_id (int | None) – The parent of the node, null if it is the root node.
parent_external_id (str | None) – The external ID of the parent. The property is omitted if the asset doesn’t have a parent or if the parent doesn’t have externalId.
description (str | None) – The description of the asset.
data_set_id (int | None) – The id of the dataset this asset belongs to.
metadata (dict[str, str] | None) – Custom, application-specific metadata. String key -> String value. Limits: Maximum length of key is 128 bytes, value 10240 bytes, up to 256 key-value pairs, of total size at most 10240.
source (str | None) – The source of the asset.
labels (list[Label | str | LabelDefinition | dict] | None) – A list of the labels associated with this resource item.
geo_location (GeoLocation | None) – The geographic metadata of the asset.
id (int | None) – A server-generated ID for the object.
created_time (int | None) – The number of milliseconds since 00:00:00 Thursday, 1 January 1970, Coordinated Universal Time (UTC), minus leap seconds.
last_updated_time (int | None) – The number of milliseconds since 00:00:00 Thursday, 1 January 1970, Coordinated Universal Time (UTC), minus leap seconds.
root_id (int | None) – ID of the root asset.
aggregates (AggregateResultItem | dict[str, Any] | None) – Aggregated metrics of the asset
cognite_client (CogniteClient | None) – The client to associate with this object.

as_write() → AssetWrite: Returns this Asset in its writing version.

children() → AssetList

Returns the children of this asset.

Returns: The requested assets
Return type: AssetList

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]

events(**kwargs: Any) → EventList

Retrieve all events related to this asset.

Parameters: **kwargs (Any) – All extra keyword arguments are passed to events/list.
Returns: All events related to this asset.
Return type: EventList

files(**kwargs: Any) → FileMetadataList

Retrieve all files metadata related to this asset.

Parameters: **kwargs (Any) – All extra keyword arguments are passed to files/list.
Returns: Metadata about all files related to this asset.
Return type: FileMetadataList

parent() → Asset

Returns this asset’s parent.

Returns: The parent asset.
Return type: Asset

sequences(**kwargs: Any) → SequenceList

Retrieve all sequences related to this asset.

Parameters: **kwargs (Any) – All extra keyword arguments are passed to sequences/list.
Returns: All sequences related to this asset.
Return type: SequenceList

subtree(depth: int | None = None) → AssetList

Returns the subtree of this asset up to a specified depth.

Parameters: depth (int | None) – Retrieve assets up to this depth below the asset.
Returns: The requested assets sorted topologically.
Return type: AssetList

time_series(**kwargs: Any) → TimeSeriesList

Retrieve all time series related to this asset.

Parameters: **kwargs (Any) – All extra keyword arguments are passed to time_series/list.
Returns: All time series related to this asset.
Return type: TimeSeriesList

to_pandas(expand_metadata: bool = False, metadata_prefix: str = 'metadata.', expand_aggregates: bool = False, aggregates_prefix: str = 'aggregates.', ignore: list[str] | None = None, camel_case: bool = False, convert_timestamps: bool = True) → pandas.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.
expand_aggregates (bool) – Expand the aggregates into separate rows (default: False).
aggregates_prefix (str) – Prefix to use for the aggregates 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

Bases: WriteableCogniteResource[AssetWrite], ABC

A representation of a physical asset, for example, a factory or a piece of equipment. This is the parent class for the Asset and AssetWrite classes.

Parameters

external_id (str | None) – The external ID provided by the client. Must be unique for the resource type.
name (str | None) – The name of the asset.
parent_id (int | None) – The parent of the node, null if it is the root node.
parent_external_id (str | None) – The external ID of the parent. The property is omitted if the asset doesn’t have a parent or if the parent doesn’t have externalId.
description (str | None) – The description of the asset.
data_set_id (int | None) – The id of the dataset this asset belongs to.
metadata (dict[str, str] | None) – Custom, application-specific metadata. String key -> String value. Limits: Maximum length of key is 128 bytes, value 10240 bytes, up to 256 key-value pairs, of total size at most 10240.
source (str | None) – The source of the asset.
labels (list[Label] | None) – A list of the labels associated with this resource item.
geo_location (GeoLocation | None) – The geographic metadata of the asset.

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]

Bases: CogniteFilter

Filter on assets with strict matching.

Parameters

name (str | None) – The name of the asset.
parent_ids (Sequence[int] | None) – Return only the direct descendants of the specified assets.
parent_external_ids (SequenceNotStr[str] | None) – Return only the direct descendants of the specified assets.
asset_subtree_ids (Sequence[dict[str, Any]] | None) – Only include assets in subtrees rooted at the specified asset IDs and external IDs. If the total size of the given subtrees exceeds 100,000 assets, an error will be returned.
data_set_ids (Sequence[dict[str, Any]] | None) – No description.
metadata (dict[str, str] | None) – Custom, application specific metadata. String key -> String value. Limits: Maximum length of key is 128 bytes, value 10240 bytes, up to 256 key-value pairs, of total size at most 10240.
source (str | None) – The source of the asset.
created_time (dict[str, Any] | TimestampRange | None) – Range between two timestamps.
last_updated_time (dict[str, Any] | TimestampRange | None) – Range between two timestamps.
root (bool | None) – Whether the filtered assets are root assets, or not. Set to True to only list root assets.
external_id_prefix (str | None) – Filter by this (case-sensitive) prefix for the external ID.
labels (LabelFilter | None) – Return only the resource matching the specified label constraints.
geo_location (GeoLocationFilter | None) – Only include files matching the specified geographic relation.

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.assets.AssetHierarchy(assets: Sequence[Asset | AssetWrite], ignore_orphans: bool = False)

Bases: object

Class that verifies if a collection of Assets is valid, by validating its internal consistency. This is done “offline”, meaning CDF is -not- queried for the already existing assets. As a result, any asset providing a parent link by ID instead of external ID, are assumed valid.

Parameters

assets (Sequence[Asset | AssetWrite]) – Sequence of assets to be inspected for validity.
ignore_orphans (bool) – If true, orphan assets are assumed valid and won’t raise.

Examples

>>> from cognite.client import CogniteClient
>>> from cognite.client.data_classes import AssetHierarchy
>>> client = CogniteClient()
>>> hierarchy = AssetHierarchy(assets)
>>> # Get a report written to the terminal listing any issues:
>>> hierarchy.validate_and_report()
>>> if hierarchy.is_valid():
...     res = client.assets.create_hierarchy(hierarchy)
... # If there are issues, you may inspect them directly:
... else:
...     hierarchy.orphans
...     hierarchy.invalid
...     hierarchy.unsure_parents
...     hierarchy.duplicates
...     hierarchy.cycles  # Requires no other basic issues

There are other ways to generate the report than to write directly to screen. You may pass an output_file which can be either a Path object (writes are done in append-mode) or a file-like object supporting write (default is None which is just regular print):

>>> # Get a report written to file:
>>> from pathlib import Path
>>> report = Path("path/to/my_report.txt")
>>> hierarchy = AssetHierarchy(assets)
>>> hierarchy.validate_and_report(output_file=report)

>>> # Get a report as text "in memory":
>>> import io
>>> with io.StringIO() as file_like:
...     hierarchy.validate_and_report(output_file=file_like)
...     report = file_like.getvalue()

count_subtree(mapping: dict[str | None, list[Asset]]) → dict[str, int]

Returns a mapping from asset external ID to the size of its subtree (children and children of children etc.).

Parameters: mapping (dict[str | None, list[Asset]]) – The mapping returned by groupby_parent_xid(). If None is passed, will be recreated (slightly expensive).
Returns: Lookup from external ID to descendant count.
Return type: dict[str, int]

groupby_parent_xid() → dict[str | None, list[Asset]]

Returns a mapping from parent external ID to a list of its direct children.

Note

If the AssetHierarchy was initialized with ignore_orphans=True, all orphans assets, if any, are returned as part of the root assets in the mapping and can be accessed by mapping[None]. The same is true for all assets linking its parent by ID.

Returns: No description.
Return type: dict[str | None, list[Asset]]

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

Bases: WriteableCogniteResourceList[AssetWrite, Asset], IdTransformerMixin

events(**kwargs: Any) → EventList

Retrieve all events related to these assets.

Parameters: **kwargs (Any) – All extra keyword arguments are passed to events/list. Note: ‘sort’, ‘partitions’ and ‘limit’ can not be used.
Returns: All events related to the assets in this AssetList.
Return type: EventList

files(**kwargs: Any) → FileMetadataList

Retrieve all files metadata related to these assets.

Parameters: **kwargs (Any) – All extra keyword arguments are passed to files/list. Note: ‘limit’ can not be used.
Returns: Metadata about all files related to the assets in this AssetList.
Return type: FileMetadataList

sequences(**kwargs: Any) → SequenceList

Retrieve all sequences related to these assets.

Parameters: **kwargs (Any) – All extra keyword arguments are passed to sequences/list. Note: ‘limit’ can not be used.
Returns: All sequences related to the assets in this AssetList.
Return type: SequenceList

time_series(**kwargs: Any) → TimeSeriesList

Retrieve all time series related to these assets.

Parameters: **kwargs (Any) – All extra keyword arguments are passed to time_series/list. Note: ‘partitions’ and ‘limit’ can not be used.
Returns: All time series related to the assets in this AssetList.
Return type: TimeSeriesList

class cognite.client.data_classes.assets.AssetProperty(value)

Bases: EnumProperty

An enumeration.

class cognite.client.data_classes.assets.AssetUpdate(id: int | None = None, external_id: str | None = None)

Bases: CogniteUpdate

Changes applied to asset

Parameters

id (int) – A server-generated ID for the object.
external_id (str) – The external ID provided by the client. Must be unique for the resource type.

Bases: AssetCore

A representation of a physical asset, for example, a factory or a piece of equipment. This is the writing version of the Asset class, and is used when inserting new assets.

Parameters

name (str) – The name of the asset.
external_id (str | None) – The external ID provided by the client. Must be unique for the resource type.
parent_id (int | None) – The parent of the node, null if it is the root node.
parent_external_id (str | None) – The external ID of the parent. The property is omitted if the asset doesn’t have a parent or if the parent doesn’t have externalId.
description (str | None) – The description of the asset.
data_set_id (int | None) – The id of the dataset this asset belongs to.
metadata (dict[str, str] | None) – Custom, application-specific metadata. String key -> String value. Limits: Maximum length of key is 128 bytes, value 10240 bytes, up to 256 key-value pairs, of total size at most 10240.
source (str | None) – The source of the asset.
labels (list[Label | str | LabelDefinitionWrite | dict] | None) – A list of the labels associated with this resource item.
geo_location (GeoLocation | None) – The geographic metadata of the asset.

as_write() → AssetWrite: Returns self.

class cognite.client.data_classes.assets.AssetWriteList(resources: Collection[Any], cognite_client: CogniteClient | None = None): Bases: CogniteResourceList[AssetWrite], ExternalIDTransformerMixin

class cognite.client.data_classes.assets.SortableAssetProperty(value)

Bases: EnumProperty

An enumeration.