cognite-sdk

import os from cognite.client import CogniteClient, AsyncCogniteClient, ClientConfig, global_config from cognite.client.credentials import OAuthClientCredentials import asyncio # It is highly recommended to use environment variables for sensitive information. # Example environment variables: # COGNITE_TENANT_ID='YOUR_TENANT_ID' # COGNITE_CLIENT_ID='YOUR_CLIENT_ID' # COGNITE_CLIENT_SECRET='YOUR_CLIENT_SECRET' # COGNITE_CLUSTER='westeurope-1' # COGNITE_PROJECT='my-cdf-project' # COGNITE_CLIENT_NAME='my-python-app' # 1. Configure the client using environment variables tenant_id = os.environ.get('COGNITE_TENANT_ID', 'your-tenant-id') client_id = os.environ.get('COGNITE_CLIENT_ID', 'your-client-id') client_secret = os.environ.get('COGNITE_CLIENT_SECRET', 'your-client-secret') cluster = os.environ.get('COGNITE_CLUSTER', 'api') # e.g., 'westeurope-1' for 'https://westeurope-1.cognitedata.com' project = os.environ.get('COGNITE_PROJECT', 'my-cdf-project') client_name = os.environ.get('COGNITE_CLIENT_NAME', 'my-python-app') base_url = f"https://{cluster}.cognitedata.com" creds = OAuthClientCredentials( token_url=f"https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/token", client_id=client_id, client_secret=client_secret, scopes=[f"{base_url}/.default"] ) client_config = ClientConfig( client_name=client_name, base_url=base_url, project=project, credentials=creds ) # Optionally set a global configuration (will be used if no config is explicitly passed) global_config.default_client_config = client_config # 2. Instantiate a synchronous client (still supported in v8) sync_client = CogniteClient() print(f"Synchronous client initialized for project: {sync_client.config.project}") # Example usage: # assets = sync_client.assets.list(limit=1) # if assets: print(f"Found asset: {assets[0].name}") # 3. Instantiate an asynchronous client (new and recommended in v8) async def main(): async_client = AsyncCogniteClient() print(f"Asynchronous client initialized for project: {async_client.config.project}") # Example usage: # tss = await async_client.time_series.list(limit=1) # if tss: print(f"Found time series: {tss[0].name}") await async_client.close() if __name__ == "__main__": asyncio.run(main())