osdu-python-client 0.4.1

Python client for OSDU services

OSDU Python Client

Python client library for OSDU services, automatically generated from OpenAPI specifications and wrapped with a handwritten facade that handles authentication, retries, partition headers, and ergonomic per-operation calls.

Installation

pip install osdu-python-client

The base install supports Community/baremetal (Keycloak client credentials) and AWS (OAuth2 client credentials) — both use only httpx which is already a core dependency. For other cloud providers, install the matching extra:

Cloud	Extra	Command
Azure / Entra ID	azure	pip install 'osdu-python-client[azure]'
GCP	gcp	pip install 'osdu-python-client[gcp]'
IBM	ibm	pip install 'osdu-python-client[ibm]'
All providers	all-csp	pip install 'osdu-python-client[all-csp]'

Python 3.13+ is required.

Quick Start

from osdu_python_client import OsduClient
from osdu_python_client.generated.search.models.query_request import QueryRequest

with OsduClient() as osdu:  # config loaded from .env
    dto = osdu.search.query_records(
        body=QueryRequest(kind="osdu:wks:master-data--Wellbore:*", query="*", limit=1)
    )
    for record in dto.results:
        print(record.additional_properties)

Each osdu.<service>.<operation>(...) call auto-binds the underlying client and data-partition-id, returns the parsed response model on 2xx, and raises OsduError on non-2xx. Call .detailed(...) on any operation to get the full Response envelope (status code, headers, parsed error model).

Available Services

Attribute	Service
crs_catalog	CRS Catalog
crs_conversion	CRS Conversion
dataset	Dataset
entitlements	Entitlements
file	File
indexer	Indexer
workflow	Ingestion Workflow Service
legal	Legal
notification	Notification
partition	Partition
policy	Policy
register	Register
schema	Schema
search	Search
seismic_ddms	Seismic DDMS
storage	Storage
unit	Unit
wellbore_ddms	Wellbore DDMS

Configuration

Configuration is loaded from environment variables or a .env file:

SERVER=https://your-osdu-instance.com
DATA_PARTITION_ID=your-partition
AUTH_PROVIDER=azure_msal
CLIENT_ID=...
AUTHORITY=https://login.microsoftonline.com/<tenant>
SCOPES=.../.default
AUTH_MODE=interactive   # interactive | device_flow | client_credentials | workload_identity

Auth Providers

Auth is pluggable via AUTH_PROVIDER. Azure MSAL is built in; other CSPs can be passed directly.

Provider	Status	Required config
azure_msal	built-in	CLIENT_ID, AUTHORITY, SCOPES, AUTH_MODE
aws_cognito	stub	bring your own TokenProvider
gcp_iam	stub	bring your own TokenProvider
ibm_iam	stub	bring your own TokenProvider

Azure MSAL flows

AUTH_MODE	When to use	Required config
interactive	Local dev / tests	client_id, authority, scopes
device_flow	Headless scripts	client_id, authority, scopes
client_credentials	CI, service-to-service	+ client_secret
workload_identity	AKS / Workload Identity Federation	scopes; optional azure_client_id, azure_tenant_id, azure_federated_token_file

Tokens are cached to .msal_token_cache.bin (override with MSAL_CACHE_PATH). The transport retries 401 once with force_refresh=True and retries 429/502/503/504 with exponential backoff honouring Retry-After.

For workload_identity the token is cached in memory because there a writeable file system may not be available. The optional parameters are typically set by the Kubernetes runtime.

Custom / bring-your-own provider

Implement the TokenProvider protocol (one method) and pass it directly:

class MyAwsProvider:
    def get_token(self, force_refresh: bool = False) -> str:
        return "..."  # return a bearer token string

osdu = OsduClient(token_provider=MyAwsProvider())

Async

import asyncio
from osdu_python_client import AsyncOsduClient
from osdu_python_client.generated.search.models.query_request import QueryRequest

async def main():
    async with AsyncOsduClient() as osdu:
        dto = await osdu.search.query_records(
            body=QueryRequest(kind="osdu:wks:master-data--Wellbore:*", query="*", limit=1)
        )
        return dto.results

asyncio.run(main())

Per-service Header Overrides

# Scope extra headers to one service
with osdu.with_headers(service="crs_conversion", **{"frame-of-reference": "units=SI"}):
    osdu.crs_conversion.convert_records(body=req)

# Or apply to all services (e.g. correlation IDs)
with osdu.with_headers(**{"x-correlation-id": correlation_id}):
    ...

The async client exposes the same helper as async with osdu.with_headers(...).

Debugging

from osdu_python_client import enable_debug_logging
enable_debug_logging()                    # transport + auth at DEBUG
enable_debug_logging(include_bodies=True) # also log truncated request/response bodies

Authorization and Cookie headers are always redacted from log output. Bodies are off by default because OSDU payloads may contain PII.

Low-level: Raw Generated Client

The generated clients can be used directly with a static token when you need full control:

from osdu_python_client.generated.entitlements.api.list_group_on_behalf_of_api import (
    list_all_partition_groups,
)
from osdu_python_client.generated.entitlements.client import AuthenticatedClient

client = AuthenticatedClient(
    base_url="https://your-osdu-instance.com/api/entitlements/v2",
    token="YOUR_ACCESS_TOKEN",
)

result = list_all_partition_groups.sync_detailed(
    client=client,
    data_partition_id="your-partition-id",
    type_="NONE",
)
if result.parsed:
    for group in result.parsed.groups:
        print(group.name, group.email)

The async variant is asyncio_detailed(...).

Development

git clone https://community.opengroup.org/osdu/platform/system/sdks/osdu-python-client.git
cd osdu-python-client
uv sync --all-extras
uv run python generate_all.py
uv run pytest tests/test_facade.py tests/test_transport.py -q

For adding services, updating OpenAPI specs, the release process, and project structure, see docs/development.md.

License

Ref. License Information