Skip to main content

Use PLACEHOLDER Cloud with Apache Airflow

Goal​

You'll wire PLACEHOLDER Cloud checkpoints into your Airflow DAGs so a failed validation blocks downstream tasks the same way any other Airflow task failure would. Instead of writing a SimpleHttpOperator + custom poller pair, you install the official provider package and use first-class operators and sensors.

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

Prereqs​

  • Apache Airflow >=2.7,<3 running somewhere — Astronomer, MWAA, self-hosted, doesn't matter.
  • Python 3.9 or newer in the Airflow worker 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​

  1. Install the provider in the Airflow worker environment.

    While the package is not yet on PyPI, install it straight from the repo:

    pip install "git+https://github.com/your-org/dq-cloud.git#subdirectory=airflow"

    For a stable install, pin to a tag or commit SHA in the URL.

    Restart the Airflow webserver and scheduler after installing — they pick up new providers at boot.

  2. 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.

  3. Create an Airflow connection.

    In the Airflow UI: Admin → Connections → +. Pick DQ Cloud as the connection type. Fill in:

    • DQ Cloud API base URL — https://placeholder.example.com (the host where your PLACEHOLDER Cloud lives, no trailing slash).
    • Personal API key (dqk_…) — paste the value from step 2.

    Save. The default connection ID is dq_cloud_default; if you create the connection under a different ID, pass it as dq_cloud_conn_id in the operator.

  4. Add the operator to a DAG.

    from datetime import datetime

    from airflow import DAG
    from dq_cloud_airflow.operators import DQCloudCheckpointOperator

    ORG = "00000000-0000-0000-0000-000000000001"
    WORKSPACE = "00000000-0000-0000-0000-000000000002"
    CHECKPOINT_ORDERS = "00000000-0000-0000-0000-0000000000aa"

    with DAG(
        "nightly_warehouse_validation",
        start_date=datetime(2026, 1, 1),
        schedule="0 2 * * *",
        catchup=False,
    ) as dag:
        validate_orders = DQCloudCheckpointOperator(
            task_id="validate_orders",
            org_id=ORG,
            workspace_id=WORKSPACE,
            checkpoint_id=CHECKPOINT_ORDERS,
        )

    The operator enqueues the checkpoint via PLACEHOLDER Cloud's run-now endpoint, polls every 10 seconds until the run reaches a terminal state, and raises AirflowException on failed or error. The full result dict lands in XCom under the task's default key, so a downstream task can read row-level detail with ti.xcom_pull(task_ids="validate_orders").

  5. Tune polling for long-running checkpoints.

    The default operator holds an Airflow worker slot while it polls. For checkpoints that take more than a few minutes, schedule the run from one task and watch with a reschedule-mode sensor in another:

    from dq_cloud_airflow.hooks import DQCloudHook
    from dq_cloud_airflow.sensors import DQCloudCheckpointResultSensor
    from airflow.decorators import task

    @task
    def kick_off_checkpoint() -> str:
        hook = DQCloudHook()
        try:
            return hook.run_checkpoint(ORG, WORKSPACE, CHECKPOINT_ORDERS)
        finally:
            hook.close()

    run_id = kick_off_checkpoint()
    wait_for_result = DQCloudCheckpointResultSensor(
        task_id="wait_for_result",
        org_id=ORG,
        workspace_id=WORKSPACE,
        run_id=run_id,
        mode="reschedule",
        poke_interval=60,
    )
    run_id >> wait_for_result

Verify​

  • The DAG run shows the task starting, the Logs tab streams DQ Cloud checkpoint run … status=running lines every 10 seconds, and the task turns green once the run reaches passed.
  • In PLACEHOLDER Cloud, a fresh ValidationResult row appears under the checkpoint's results list, attributed to the personal API key you used.
  • Pull the XCom: ti.xcom_pull(task_ids="validate_orders") returns the full result dict (status, success, rule-level breakdown).

Troubleshooting​

  • AirflowException: 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 workspace_id in the operator.
  • AirflowException: missing 'host' — the Airflow connection was saved without a base URL. Re-open it, paste the URL, save.
  • Task hangs at status=pending — PLACEHOLDER Cloud accepted the run but no worker has picked it up. Check the PLACEHOLDER Cloud worker logs. If the checkpoint targets an agent-mode datasource, confirm the on-prem agent is connected.
  • Task fails with did not reach a terminal status within Ns — bump the operator's timeout argument, or move to the sensor-based pattern in step 5 so polling doesn't hold a worker slot.

Out of scope​

  • A @task-style TaskFlow decorator wrapper is a follow-up — see the FEATURE-26 notes.
  • A dbt operator lives in FEATURE-27, tracked separately.