Use Measured Cloud with Dagster

Goal

You'll model Measured Cloud validations as Dagster asset checks so a failed validation shows up in the asset's reliability sidebar — the shift-left, asset-based story — instead of in a separate dashboard. Instead of hand-rolling an op that POSTs to the API and polls for a result, you install the official provider package and register checks with one line each.

The provider package source lives at providers/dagster/ in this repo and ships under the name dagster-dq-cloud (#86). Publishing to PyPI is on the roadmap; until then, install directly from the repo (see below).

Coming from checkpoints? Checkpoints were retired (see Checkpoints (retired)). dq_cloud_check(checkpoint_id=...), DQCloudResource.run_checkpoint and checkpoints_to_asset_checks still work for one release as deprecated shims: they run the validation the checkpoint maps to and emit a DeprecationWarning. Per-run rule_parameters overrides are no longer applied. Switch to validation_id= — the shims are removed in the next major release.

Prereqs

Dagster >=1.7 running somewhere — Dagster OSS / self-managed, Dagster+, or a local dagster dev.
Python 3.9 or newer in the Dagster environment.
A Measured Cloud organisation + workspace + at least one validation to run. Every data asset with rules has a default validation, so to check one table, run its default validation.
Workspace editor role or higher in Measured Cloud (so your personal API key can trigger runs).

Steps

Install the provider in the Dagster environment.

While the package is not yet on PyPI, install it straight from the repo:
```
pip install "git+https://github.com/tomplace/DQ-Cloud.git#subdirectory=providers/dagster"
```
For a stable install, pin to a tag or commit SHA in the URL.
Generate a personal API key in Measured Cloud.

In the UI, open Settings → API keys → New API key (see Create an API token). Copy the dqk_… value somewhere safe — you only see it once. Export it (and your org / workspace ids) to the environment so they stay out of code:
```
export DQ_CLOUD_API_KEY="dqk_…"
export DQ_CLOUD_ORG_ID="00000000-0000-0000-0000-000000000001"
export DQ_CLOUD_WORKSPACE_ID="00000000-0000-0000-0000-000000000002"
```

Wire the resource into your Definitions.

DQCloudResource is a Dagster ConfigurableResource. Set it up once and use EnvVar so the API key never lands in code or the Definitions snapshot.

from dagster import Definitions, EnvVar
from dagster_dq_cloud import DQCloudResource

dq = DQCloudResource(
    base_url="https://dq.example.com",      # no trailing slash
    api_key=EnvVar("DQ_CLOUD_API_KEY"),               # dqk_… personal API key
    org_id=EnvVar("DQ_CLOUD_ORG_ID"),
    workspace_id=EnvVar("DQ_CLOUD_WORKSPACE_ID"),
)

Register an asset check with dq_cloud_check.

The factory collapses the "start a validation run → poll → emit AssetCheckResult" pattern into one line.
```
from dagster import AssetKey, Definitions, asset
from dagster_dq_cloud import DQCloudResource, dq_cloud_check

VALIDATION_ORDERS = "00000000-0000-0000-0000-0000000000aa"

@asset
def orders() -> int:
    ...

orders_dq = dq_cloud_check(
    validation_id=VALIDATION_ORDERS,
    asset=AssetKey(["orders"]),
    name="orders_dq",      # optional — defaults to dq_cloud_<short_id>
)

defs = Definitions(
    assets=[orders],
    asset_checks=[orders_dq],
    resources={"dq": dq},
)
```
When Dagster materialises the check, the provider starts an execution run of the validation (POST /api/v2/.../validations/{id}/run), polls it until it finishes, and returns an AssetCheckResult with passed set when the run succeeded, plus metadata for rules_evaluated, rules_failed, status, health_score, execution_run_id, validation_id, and a deep-link validation_url. Pass rule_ids=[...] to run only some of the validation's rules. A failed validation surfaces on the asset's check sidebar; pass blocking=True if you also want it to fail the upstream materialisation.
(Deprecated) Auto-discover checks from the checkpoint catalog.

checkpoints_to_asset_checks reads the retired checkpoint catalog and emits one check per checkpoint that targets a configured asset; each check runs the validation its checkpoint maps to. It is kept for one release. Prefer explicit dq_cloud_check(validation_id=...) lines for new code. The caller supplies the bridge between Dagster asset keys and cloud data_asset_ids — the most ergonomic form stashes the cloud id in the asset's metadata:
```
from dagster import Definitions, asset
from dagster_dq_cloud import checkpoints_to_asset_checks

@asset(metadata={"dq_cloud_asset_id": "44444444-4444-4444-4444-444444444444"})
def orders() -> int:
    ...

checks = checkpoints_to_asset_checks(dq, assets=[orders])

defs = Definitions(
    assets=[orders],
    asset_checks=checks,
    resources={"dq": dq},
)
```
A bare asset with no dq_cloud_asset_id (in a tuple or in metadata) raises ValueError at definition-load time, so a forgotten mapping fails loudly instead of silently dropping checks.

Verify

In the Dagster UI, the asset's Checks tab lists orders_dq. Run it — the check turns green when the validation run succeeds, red when it fails, with the failed-rule count in the metadata.
The check metadata includes a validation_url that deep-links to the run in Measured Cloud.
In Measured Cloud, a new run appears on the validation's Execution runs card (trigger api).

Troubleshooting

DQCloudClientError: … returned 401 — the dqk_… key was rejected. Confirm it hasn't been revoked in Settings → API keys, and that the workspace it's scoped to matches the resource's workspace_id.
DQCloudClientError: run_validation response missing execution_run_id — the API accepted the request but returned an unexpected ack shape. Confirm the base_url points at the API host (not the UI) and that the validation id exists in the configured workspace.
TimeoutError: … did not reach a terminal state within Ns — Measured Cloud accepted the run but it didn't finish in time. Bump run_timeout_seconds on the resource, or check the worker logs. If the validation includes an agent-mode datasource, confirm the on-prem agent is connected: the run stays running until the agent reports.
ValueError: … without a dq_cloud_asset_id — an asset passed to checkpoints_to_asset_checks carries no cloud asset id. Add a (asset, dq_cloud_asset_id) tuple or a dq_cloud_asset_id metadata entry.

Out of scope

The Dagster+ / Dagster Cloud OAuth flow — the recipe above assumes a self-managed Dagster with a personal API key.

Goal​

Prereqs​

Steps​

Verify​

Troubleshooting​

Out of scope​