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 data sources, running a validation, and requesting an AI profile run — without writing any HTTP code yourself.

Prereqs​

  • Python 3.12 or newer.
  • A running Measured 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 validation and wait​

Validations are exposed as client.validations, and their runs as client.execution_runs:

from dq_cloud import FAILED_STATUSES, DQCloudClient, DQCloudError

with DQCloudClient(api_key="dqk_...", base_url="https://dq.example.com") as client:
    try:
        run = client.validations.run(
            "validation-uuid",
            org_id="org-uuid",
            workspace_id="ws-uuid",
            wait=True,
            timeout=300.0,
        )
    except DQCloudError as exc:
        print("could not start the run:", exc)
        raise

    print("status:", run["status"], "health:", run.get("health_score"))
    if run["status"] in FAILED_STATUSES:
        raise SystemExit(1)

Without wait=True the call returns as soon as the run is queued, with its execution_run_id and parent_result_id. Follow it later with client.execution_runs.wait(execution_run_id, org_id=..., workspace_id=...), or read its per-asset and per-rule breakdown with client.execution_runs.results(...). wait=True polls the execution run every 2 seconds (change it with poll_interval=...) and raises TimeoutError after timeout seconds; a failed run is returned, not raised, so check its status. Optional arguments: rule_ids=[...] runs only some of the validation's rules, and batch_definition_id with partition_value (or partition_values, one run per value) runs one partition. Stop a run with client.validations.cancel_run(validation_id, parent_result_id=..., org_id=...). A validation with nothing to run is rejected with a 409.

To check one table, run its default validation: every data asset with rules has one.

Deprecated: client.checkpoints. Checkpoints were retired (see Checkpoints (retired)). client.checkpoints still works for one release and emits a DeprecationWarning: list / get return the legacy rows with their mapped validation_id, and run runs that validation and returns its execution_run_id (with wait=True, the finished execution run). create / update / delete raise CheckpointsRetiredError. It is removed in the next major release — switch to client.validations.

Example 3 — add a rule to an asset​

Rules are exposed as client.rules. A rule lives on its data asset; it always joins the asset's default validation, and validation_ids adds it to other validations too:

from dq_cloud import DQCloudClient

with DQCloudClient(api_key="dqk_...", base_url="https://dq.example.com") as client:
    rule = client.rules.create(
        "asset-uuid",
        org_id="org-uuid",
        workspace_id="ws-uuid",
        rule_type="expect_column_values_to_not_be_null",
        kwargs={"column": "order_id"},
        validation_ids=["validation-uuid"],
    )
    for r in client.rules.list(org_id="org-uuid", workspace_id="ws-uuid",
                               validation_id="validation-uuid"):
        print(r["data_asset_id"], r["rule_type"], r["kwargs"])
    run = client.rules.run("asset-uuid", rule["id"], org_id="org-uuid",
                           workspace_id="ws-uuid", wait=True)

client.rules.list filters by data_asset_id or validation_id (list_page pages with a cursor), get returns one rule with the validation_ids that hold it, update edits a rule in place (it keeps its id), delete removes it, and set_validations sets which validations hold it. An invalid rule (a bad custom-SQL rule, or a row_condition that isn't Great Expectations syntax) raises ValidationFailedError.

Deprecated: client.rule_suites and client.rule_groups. Rule suites were retired (see Rule suites (retired)). For one release client.rule_suites reads the API's read-only alias (list, get, validate), and its writes raise RuleSuitesRetiredError. client.rule_groups is the old name of client.validations; its add_member / remove_member (the retired asset list) raise ValidationMembersRetiredError. Every call on either emits a DeprecationWarning. Switch to client.rules and client.validations.

Example 4 — 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.