Use PLACEHOLDER Cloud with Dagster

Goal

You'll model PLACEHOLDER Cloud checkpoints 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).

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 PLACEHOLDER Cloud organisation + workspace + at least one checkpoint to call.
Workspace editor role or higher in PLACEHOLDER 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 PLACEHOLDER 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://placeholder.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 "enqueue checkpoint → poll → emit AssetCheckResult" pattern into one line.
```
from dagster import AssetKey, Definitions, asset
from dagster_dq_cloud import DQCloudResource, dq_cloud_check

CHECKPOINT_ORDERS = "00000000-0000-0000-0000-0000000000aa"

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

orders_dq = dq_cloud_check(
    checkpoint_id=CHECKPOINT_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 POSTs to the run-now endpoint, polls run-queue/{id} until the run reaches a terminal state, and returns an AssetCheckResult with passed=result.success plus metadata for expectations_evaluated, expectations_failed, status, and a deep-link checkpoint_url. A failed validation surfaces on the asset's check sidebar; pass blocking=True if you also want it to fail the upstream materialisation.
(Optional) Auto-discover checks from the checkpoint catalog.

If you've already organised many assets and checkpoints in PLACEHOLDER Cloud, checkpoints_to_asset_checks queries the catalog and emits one check per match. 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 checkpoint passes, red when it fails, with the failed-expectation count in the metadata.
The check metadata includes a checkpoint_url that deep-links to the ValidationResult in PLACEHOLDER Cloud.
In PLACEHOLDER Cloud, a fresh ValidationResult row appears under the checkpoint's results list, attributed to the personal API key you used.

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_checkpoint response missing run_queue_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 checkpoint id exists in the configured workspace.
TimeoutError: … did not reach a terminal state within Ns — PLACEHOLDER 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 checkpoint targets an agent-mode datasource, confirm the on-prem agent is connected.
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.

Tip: Per-run rule_parameters overrides are live. Pass rule_parameters={...} to run_checkpoint (or to the dq_cloud_check factory) and the server merges them on top of the checkpoint's stored parameters for that single run — useful for relaxing a threshold during a backfill without editing the checkpoint config.

Goal​

Prereqs​

Steps​

Verify​

Troubleshooting​

Out of scope​