Skip to main content

Use the Python SDK

Goal​

You'll install the official dq-cloud Python SDK, authenticate with a personal API key, and run three worked examples — listing datasources, running a checkpoint, and requesting an AI profile run — without writing any HTTP code yourself.

Prereqs​

  • Python 3.12 or newer.
  • A running PLACEHOLDER Cloud instance and an org you can see in the UI.
  • A personal API key — mint one under Settings → API keys (see Create an API token). The key starts with dqk_. Agent tokens (dqa_) work too; JWT session tokens do not.

Install​

pip install dq-cloud

The package targets Python 3.12+. It depends on httpx and re-exports the same Pydantic models the API uses (dq-shared), so request and response shapes stay typed end-to-end.

Authenticate​

The SDK takes the API key and the base URL of your install:

from dq_cloud import DQCloudClient

client = DQCloudClient(
    api_key="dqk_...",
    base_url="https://dq.example.com",
)

Use it as a context manager when you want the underlying connection pool closed cleanly:

with DQCloudClient(api_key="dqk_...", base_url="https://dq.example.com") as client:
    ...

429 and 503 responses are retried automatically with exponential backoff (plus jitter). Pass max_retries=0 to opt out, or tune backoff_base / backoff_cap to fit your environment.

Example 1 — list datasources​

from dq_cloud import DQCloudClient

org_id = "11111111-1111-1111-1111-111111111111"

with DQCloudClient(api_key="dqk_...", base_url="https://dq.example.com") as client:
    for ds in client.datasources.list(org_id=org_id):
        print(ds["name"], ds["type"], ds["execution_mode"])

If the key is wrong you'll see a dq_cloud.AuthError. If the org doesn't exist (or your key isn't scoped to it) you'll see dq_cloud.NotFoundError.

Example 2 — run a checkpoint and wait​

from dq_cloud import DQCloudClient, DQCloudError

with DQCloudClient(api_key="dqk_...", base_url="https://dq.example.com") as client:
    try:
        result = client.checkpoints.run(
            "checkpoint-uuid",
            org_id="org-uuid",
            workspace_id="ws-uuid",
            wait=True,
            timeout=300.0,
        )
    except DQCloudError as exc:
        print("checkpoint failed:", exc)
        raise

    print("success?", result["success"])
    print("ran at", result["run_time"])

wait=True polls validation-results until the worker writes a terminal row. The default poll interval is 2 seconds — change it with poll_interval=.... If the checkpoint takes longer than timeout seconds the SDK raises TimeoutError.

Example 3 — request an AI profile​

import time
from dq_cloud import DQCloudClient, ProfileRunResult

with DQCloudClient(api_key="dqk_...", base_url="https://dq.example.com") as client:
    queued = client.ai.request_profile(
        org_id="org-uuid",
        workspace_id="ws-uuid",
        data_asset_id="asset-uuid",
    )
    run_id = queued["id"]

    while True:
        raw = client.ai.get_profile(run_id, org_id="org-uuid", workspace_id="ws-uuid")
        profile = ProfileRunResult.model_validate(raw)
        if profile.status in ("completed", "failed"):
            break
        time.sleep(2)

    for col in profile.columns:
        print(col.name, col.dtype, f"{col.null_rate:.2%}")

ProfileRunResult (and every other re-exported model under dq_cloud.models) is the same Pydantic class the API uses, so attribute access is type-checked by Pyright / Mypy.

Verify​

  • pip show dq-cloud reports a version starting at 0.1.0.
  • python examples/01_list_datasources.py (in the SDK repo) prints your datasources.
  • Errors raise typed exceptions (AuthError, NotFoundError, RateLimitError, ValidationFailedError, ServerError) — your except clauses don't need to inspect response.status_code.

Caveats​

  • The SDK is sync-first. There's no async variant yet — wrap calls in asyncio.to_thread if you need to fan out from an async caller.
  • Streaming endpoints (data-docs blobs, large export downloads) aren't exposed. Use a plain httpx call for those for now.
  • Token revocation isn't instant on the server (HARDEN-7 cache). A revoked key may keep working for up to TOKEN_CACHE_TTL_SECONDS (default 5 seconds) after revocation.