Create

async AsyncCogniteClient.transformations.external_data_sources.create(

data_source: ExternalDataSourceWrite | Sequence[ExternalDataSourceWrite],

) → ExternalDataSource | ExternalDataSourceList

Create external data sources.

Both OneLake and Snowflake data sources can be created, also in the same call. For Snowflake, the client sends no keys: the server generates the key pair and only the public key is returned, which you must register with the Snowflake user. expiry_time is required for Snowflake and not allowed for OneLake.

Fails with a 409 if any external ID in the request already exists for the project; no items are created in that case. To modify an existing data source, delete it and create a replacement.

Parameters:

data_source (ExternalDataSourceWrite | Sequence[ExternalDataSourceWrite]) – The data source(s) to create.

Returns:

The created data source(s).

Return type:

ExternalDataSource | ExternalDataSourceList

Raises:

CogniteDuplicatedError – If an external ID in the request already exists for the project.

Examples

Register a Fabric OneLake data source:

>>> import os
>>> from cognite.client import CogniteClient, AsyncCogniteClient
>>> from cognite.client.data_classes.transformations.externaldata import (
...     OneLakeCredentialsWrite,
...     OneLakeExternalDataSourceWrite,
...     OneLakeLocationDescription,
...     OneLakeSettingsWrite,
... )
>>> client = CogniteClient()
>>> # async_client = AsyncCogniteClient()  # another option
>>> data_source = OneLakeExternalDataSourceWrite(
...     external_id="fabric-lakehouse-prod",
...     name="Production lakehouse",
...     data_set_id=123456,
...     settings=OneLakeSettingsWrite(
...         credentials=OneLakeCredentialsWrite(
...             client_id="<azure-app-id>",
...             tenant_id="<azure-tenant-uuid>",
...             client_secret=os.environ["ONELAKE_CLIENT_SECRET"],
...         ),
...         location_description=OneLakeLocationDescription(
...             workspace_id="<fabric-workspace-guid>",
...             container_id="<fabric-lakehouse-guid>",
...         ),
...     ),
... )
>>> res = client.transformations.external_data_sources.create(data_source)

Register a Snowflake data source, valid for 30 days:

>>> from cognite.client.data_classes.transformations.externaldata import (
...     SnowflakeCredentialsWrite,
...     SnowflakeExternalDataSourceWrite,
...     SnowflakeLocationDescription,
...     SnowflakeSettingsWrite,
... )
>>> from cognite.client.utils._time import timestamp_to_ms
>>> snowflake_source = SnowflakeExternalDataSourceWrite(
...     external_id="snowflake-analytics-prod",
...     expiry_time=timestamp_to_ms("30d-ahead"),
...     settings=SnowflakeSettingsWrite(
...         credentials=SnowflakeCredentialsWrite(
...             account_identifier="org-account",
...             user_name="COGNITE_SVC",
...             role_name="COGNITE_READER",
...         ),
...         location_description=SnowflakeLocationDescription(
...             warehouse_name="COMPUTE_WH"
...         ),
...     ),
... )
>>> created = client.transformations.external_data_sources.create(snowflake_source)
>>> public_key = (
...     created.settings.credentials.public_key
... )  # register with the Snowflake user