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

AssetsAPI.list(name: str | None = None, parent_ids: Sequence[int] | None = None, parent_external_ids: SequenceNotStr[str] | None = None, asset_subtree_ids: int | Sequence[int] | None = None, asset_subtree_external_ids: str | SequenceNotStr[str] | None = None, data_set_ids: int | Sequence[int] | None = None, data_set_external_ids: str | SequenceNotStr[str] | None = None, labels: LabelFilter | None = None, geo_location: GeoLocationFilter | None = None, metadata: dict[str, str] | None = None, source: str | None = None, created_time: dict[str, Any] | TimestampRange | None = None, last_updated_time: dict[str, Any] | TimestampRange | None = None, root: bool | None = None, external_id_prefix: str | None = None, aggregated_properties: Sequence[AggregateAssetProperty] | None = None, partitions: int | None = None, limit: int | None = 25) AssetList

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) – Asset subtree id or list of asset subtree ids to filter on.

  • asset_subtree_external_ids (str | SequenceNotStr[str] | None) – Asset subtree external id or list of asset subtree external ids to filter on.

  • 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 assets in parallel using this number of workers. Also requires limit=None to be passed. To prevent unexpected problems and maximize read throughput, API documentation recommends at most use 10 partitions. When using more than 10 partitions, actual throughout decreases. In future releases of the APIs, CDF may reject requests with more than 10 partitions.

  • limit (int | None) – Maximum number of assets to return. Defaults to 25. Set to -1, float(“inf”) or None to return all items.

Returns

List of requested assets

Return type

AssetList

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)

Aggregate assets

AssetsAPI.aggregate(filter: AssetFilter | dict | None = None) list[CountAggregate]

Aggregate assets

Parameters

filter (AssetFilter | dict | 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

AssetsAPI.aggregate_count(property: AssetPropertyLike | None = None, advanced_filter: Filter | dict | None = None, filter: AssetFilter | dict | None = None) int

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 | None) – The advanced filter to narrow down the assets to count.

  • filter (AssetFilter | dict | 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

AssetsAPI.aggregate_cardinality_values(property: AssetPropertyLike, advanced_filter: Filter | dict | None = None, aggregate_filter: AggregationFilter | dict | None = None, filter: AssetFilter | dict | None = None) int

Find approximate property count for assets.

Parameters
  • property (AssetPropertyLike) – The property to count the cardinality of.

  • advanced_filter (Filter | dict | None) – The advanced filter to narrow down assets.

  • aggregate_filter (AggregationFilter | dict | None) – The filter to apply to the resulting buckets.

  • filter (AssetFilter | dict | 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

AssetsAPI.aggregate_cardinality_properties(path: AssetPropertyLike, advanced_filter: Filter | dict | None = None, aggregate_filter: AggregationFilter | dict | None = None, filter: AssetFilter | dict | None = None) int

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 | None) – The advanced filter to narrow down assets.

  • aggregate_filter (AggregationFilter | dict | None) – The filter to apply to the resulting buckets.

  • filter (AssetFilter | dict | 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

AssetsAPI.aggregate_unique_values(property: AssetPropertyLike, advanced_filter: Filter | dict | None = None, aggregate_filter: AggregationFilter | dict | None = None, filter: AssetFilter | dict | None = None) UniqueResultList

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 | None) – The advanced filter to narrow down assets.

  • aggregate_filter (AggregationFilter | dict | None) – The filter to apply to the resulting buckets.

  • filter (AssetFilter | dict | 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

AssetsAPI.aggregate_unique_properties(path: AssetPropertyLike, advanced_filter: Filter | dict | None = None, aggregate_filter: AggregationFilter | dict | None = None, filter: AssetFilter | dict | None = None) UniqueResultList

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 | None) – The advanced filter to narrow down assets.

  • aggregate_filter (AggregationFilter | dict | None) – The filter to apply to the resulting buckets.

  • filter (AssetFilter | dict | 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

AssetsAPI.search(name: str | None = None, description: str | None = None, query: str | None = None, filter: AssetFilter | dict | None = None, limit: int = 25) AssetList

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 | 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:

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

    • Missing external ID.

    • Missing a valid name.

    • Has an ID set.

  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 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

AssetsAPI.filter(filter: Filter | dict, sort: SortSpec | list[SortSpec] | None = None, aggregated_properties: Sequence[AggregateAssetProperty] | None = None, limit: int | None = 25) AssetList

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 as flt
>>> client = CogniteClient()
>>> in_timezone = flt.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 as flt
>>> from cognite.client.data_classes.assets import AssetProperty, SortableAssetProperty
>>> client = CogniteClient()
>>> in_timezone = flt.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.

class cognite.client.data_classes.assets.Asset(external_id: str | None = None, name: str | None = None, parent_id: int | None = None, parent_external_id: str | None = None, description: str | None = None, data_set_id: int | None = None, metadata: dict[str, str] | None = None, source: str | None = None, labels: list[Label | str | LabelDefinition | dict] | None = None, geo_location: GeoLocation | None = None, id: int | None = None, created_time: int | None = None, last_updated_time: int | None = None, root_id: int | None = None, aggregates: AggregateResultItem | dict[str, Any] | None = None, cognite_client: CogniteClient | None = None)

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

class cognite.client.data_classes.assets.AssetCore(external_id: str | None = None, name: str | None = None, parent_id: int | None = None, parent_external_id: str | None = None, description: str | None = None, data_set_id: int | None = None, metadata: dict[str, str] | None = None, source: str | None = None, labels: list[Label] | None = None, geo_location: GeoLocation | None = None)

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]

class cognite.client.data_classes.assets.AssetFilter(name: str | None = None, parent_ids: Sequence[int] | None = None, parent_external_ids: SequenceNotStr[str] | None = None, asset_subtree_ids: Sequence[dict[str, Any]] | None = None, data_set_ids: Sequence[dict[str, Any]] | None = None, metadata: dict[str, str] | None = None, source: str | None = None, created_time: dict[str, Any] | TimestampRange | None = None, last_updated_time: dict[str, Any] | TimestampRange | None = None, root: bool | None = None, external_id_prefix: str | None = None, labels: LabelFilter | None = None, geo_location: GeoLocationFilter | None = None)

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 assets (including the roots given). 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 | AssetWrite]]) 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 | AssetWrite]]) – 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 | AssetWrite]]

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 | AssetWrite]]

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.

class cognite.client.data_classes.assets.AssetWrite(name: str, external_id: str | None = None, parent_id: int | None = None, parent_external_id: str | None = None, description: str | None = None, data_set_id: int | None = None, metadata: dict[str, str] | None = None, source: str | None = None, labels: list[Label | str | LabelDefinitionWrite | dict] | None = None, geo_location: GeoLocation | None = None)

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.