airflow-dq 0.2.0

Airflow-3-native data-quality & data-contract plugin with an integrated quality dashboard

airflow-dq

An Airflow-3-native data-quality & data-contract plugin with an integrated quality dashboard.

You author ODCS data contracts inside the Airflow UI, they compile into real check-tasks in your DAG, and the results land in a built-in dashboard — no external SaaS backend, no separate platform.

Status: 🟢 v0.2.0 — feature-complete beta. 327 unit tests, end-to-end integration test in CI, built UI ships in the wheel. Not yet battle-tested in a multi-team production deployment.

Licensed under Apache-2.0 · Changelog · Contributing · Security policy. airflow-dq is an independent community project — it is not endorsed by or affiliated with the Apache Software Foundation or the Apache Airflow project (portions of the UI are derived from Apache Airflow, see NOTICE).

---

Why

Data-quality tooling (Soda, Great Expectations, Elementary, Monte Carlo) is mostly checks-as-code in YAML files next to the pipeline, often backed by external SaaS. None of them give you contract-native DQ that lives entirely inside the orchestrator: visual authoring → codegen to real tasks → integrated dashboard, as pure self-hosted OSS.

Features

• 12 rule types — notNull, unique, valueRange, acceptedValues, pattern (regex), referentialIntegrity, freshness (max-age or row-age, day/hour units, DATE + TIMESTAMP), rowCount, schema drift, custom SQL expression, z-score anomaly, cross-table reconciliation — all compiled to push-down SQL with a single failing predicate feeding count, bad-row sample, and quarantine.
• ODCS v3.1 SLA checks — slaProperties (latency, frequency, retention) compile into sla.* checks with the same gate/monitor semantics; unsupported properties are surfaced as compile warnings.
• Gate / monitor / quarantine — one contract runs as an inline blocking gate (on_fail="fail"), a record-only asset-triggered monitor (on_fail="warn"), or a bad-row quarantine (on_fail="quarantine"), fanned out via Dynamic Task Mapping into one visible task per check.
• Quarantine triage UI — browse dq.dq_quarantine (filter by dataset/check/state), inspect quarantined rows, mark them resolved. Quarantine writes are retry-safe (replace-on-retry by run key).
• Overview dashboard — per-contract portfolio view (pass rate, failing checks, last run, SLA status, anomaly count) plus the per-contract dashboard: KPI row, pass-rate trend, per-dimension breakdown, expandable bad-row samples, deep links to the Airflow run.
• Anomaly detection, two levels — row-level z-score outliers (with baselineWhere to keep the baseline honest), and server-side metric anomalies: each run's failing-row count is tested against a trailing window (excluding the run itself), persisted to dq.dq_metric_anomalies, flagged on the trend chart, and alertable.
• Breaking-change detection + PR propose flow — every save diffs against the previous version (column removed / type changed / new required / uniqueness added = breaking); versions are immutable (server-side numbering, HTTP 409 on conflict, atomic writes) and mirrored into a registry (dq_contracts / dq_contract_versions). With DQ_CONTRACTS_GIT=true, "propose" commits the new version on a dq/contract-<id>-v<N> branch for PR review instead of writing to the tree.
• dbt import — convert an existing dbt schema.yml (not_null / unique / accepted_values / relationships) into a contract draft via POST /dq/api/contracts/import/dbt.
• Auth via the Airflow auth manager — API routes validate Airflow-issued JWTs (header or the Airflow UI session cookie); no token is ever injected into served HTML. A legacy shared secret (DQ_API_TOKEN) remains supported for machine clients (CI, curl).
• 6 warehouse backends — Postgres and DuckDB (stable), Snowflake, BigQuery, Databricks and Trino (experimental: SQL generation is unit-tested, live connections are not). Backends resolve as Airflow Connections (conn_id == backend) with env-DSN fallbacks; pooled engines, statement timeouts.
• Alerting with fatigue controls — one grouped alert per (run, contract), retry-safe, for fail and error statuses; channels selected by route URL scheme: Slack-compatible webhook, mailto: (SMTP), pagerduty:// (Events v2), opsgenie://; per-owner and per-severity routing via DQ_ALERT_ROUTES.
• OpenLineage emission — DataQualityAssertions + metrics facets per check, best-effort, with spec-compliant dataset naming (namespace derived from the executor's connection, e.g. postgres://host:5432).
• Operability — versioned SQL migrations applied automatically (dq.dq_schema_version), error status for infra failures (monitors don't crash on a store outage), idempotent results/quarantine/alert writes keyed by (dag_id, run_id, task_id, check_name), paginated + filterable results API, optional retention (DQ_RESULTS_RETENTION_DAYS).

Architecture

flowchart TD
    A[Contract Editor<br/>React view in Airflow UI] -->|409-safe save + diff| B[Contract Store<br/>immutable ODCS-YAML versions + registry DB]
    B -->|version diff| C[Breaking-Change Detector]
    B -->|propose=true| P[Git branch<br/>PR approval flow]
    D[Your DAG] -->|build_contract_checks| E[Contract Compiler<br/>rules + SLA properties]
    B --> E
    E -->|Dynamic Task Mapping| F[One check task per rule]
    F --> G[CheckExecutor<br/>Airflow Connection or DSN]
    G -->|Postgres / DuckDB / Snowflake* / BigQuery* / Databricks* / Trino*| H[(Warehouse)]
    G -->|idempotent write| I[(Results Store<br/>dq schema + migrations)]
    G -->|best-effort| J[OpenLineage facets]
    G -->|grouped, deduped| K[Alerts<br/>webhook / email / PagerDuty / Opsgenie]
    I --> L[FastAPI plugin /dq<br/>auth: Airflow JWT or legacy token]
    L --> M[Overview · Dashboard · Quarantine Triage · Editor<br/>React views in Airflow UI]

* experimental backends

Dataflow in one sentence: author a contract → it is stored as an immutable version → a DAG calls build_contract_checks() → the compiler turns rules + SLAs into mapped check-tasks → the executor runs push-down SQL against the backend → results (plus samples/quarantine/anomalies) land idempotently in the results store and, on failure, one grouped alert goes to the owner's channel → the embedded dashboard reads the store.

Repository layout

airflow-dq/
├── src/airflow_dq/        # the installable Python package (the actual product)
│   ├── contracts/           # ODCS v3.1 models, ContractStore, registry, dbt import, git flow, diffing
│   ├── compiler/            # Contract -> CheckSpec[] (rules + SLA properties)
│   ├── executors/           # pluggable CheckExecutor + dialect layer (postgres, duckdb, snowflake*, ...)
│   ├── operators/           # DataContractCheckOperator + build_contract_checks() factory
│   ├── results/             # ResultsStore + SQL migrations (dq schema, own Postgres)
│   ├── alerting/            # grouped notifier + webhook/email/PagerDuty/Opsgenie channels
│   └── plugin/              # Airflow plugin: FastAPI app (api/) + auth + served React UI
├── ui/                      # React + TS: overview, dashboard, quarantine triage, contract editor
├── dags/                    # demo medallion DAG (gate) + asset-triggered monitor
├── contracts/               # versioned ODCS YAML (source of truth for the demo)
├── seeds/                   # intentionally "dirty" demo data
├── tests/                   # 327 unit tests (compiler/executors/store/API/operator/alerting/...)
├── scripts/                 # integration test used by CI and `make integration`
├── docker-compose.yaml      # Airflow 3 + Postgres (results + demo data)
└── DESIGN.md                # decisions & tradeoffs

Quickstart

cp .env.example .env
docker compose up -d                 # Airflow 3 + Postgres
make ui-build                        # build the React views (served by the plugin)
make test                            # unit tests

# Airflow UI:        http://localhost:8080
# Quality Dashboard: http://localhost:8080/dq/app/   (after a demo_medallion run)
# Health:            http://localhost:8080/dq/api/health

Useful targets: make integration (full compose-up → DAG run → results assertion → teardown), make db-migrate (apply pending schema migrations to DQ_RESULTS_DSN), make reseed (reload the dirty demo data so demo outcomes stop drifting with volume age), make clean.

Learn by example: examples/README.md is a 15-minute guided tour — five runnable example DAGs (dags/examples/, auto-loaded into the demo stack) covering the quickstart gate, quarantine + triage, incremental/partition-scoped checks, SLA + anomaly detection, and asset-triggered monitoring, plus dbt import, alert routing and the PR flow.

Using it in a DAG

from airflow_dq import build_contract_checks

validate_silver = build_contract_checks(
    contract="orders@v3",        # pinned, immutable version
    backend="postgres_dwh",      # Airflow conn_id (env-DSN fallback)
    on_fail="quarantine",        # "fail" | "warn" | "quarantine"
    retries=1,                   # task tuning passes through
)
bronze >> silver >> validate_silver >> gold

Two deployment patterns from the same contract:

• Gate — inline in the producer DAG (demo_medallion): checks run at the layer boundary and can block / quarantine before data flows downstream.
• Monitor — asset-triggered (dq_monitor): the table is an Airflow Asset; the monitor DAG schedule=[orders_asset] re-runs the contract whenever the producer emits the asset — on_fail="warn", record-only. Airflow 3 data-aware scheduling.

Both write to the same results store, so the dashboard reflects gate and monitor runs.

Configuration reference

All knobs are read through airflow_dq.config.settings() (src/airflow_dq/config.py). See .env.example for a commented template.

Env var	Default	Purpose
DQ_RESULTS_DSN	compose demo DSN	SQLAlchemy DSN of the DQ results/registry store (schema dq, never Airflow's metadata DB)
DQ_DWH_DSN	compose demo DSN	Fallback warehouse DSN for postgres* backends when no Airflow Connection matches
DQ_DUCKDB_PATH	:memory:	DuckDB database path for duckdb* backends
DQ_CONTRACTS_DIR	contracts	Directory of versioned contract YAML (shared by API server + workers)
DQ_SNOWFLAKE_DSN	unset	Snowflake DSN fallback (experimental adapter)
DQ_BIGQUERY_DSN	unset	BigQuery DSN fallback (experimental adapter)
DQ_DATABRICKS_DSN	unset	Databricks DSN fallback (experimental adapter)
DQ_TRINO_DSN	unset	Trino DSN fallback (experimental adapter)
DQ_API_TOKEN	unset	Legacy shared secret for machine clients; unset = auth-manager JWT/cookie only
DQ_UI_DIST	unset	Override for the built React bundle location (wheel/checkout auto-detected)
DQ_BASE_URL	empty	Absolute base URL for clickable links in alerts
DQ_STATEMENT_TIMEOUT_S	300	Statement timeout for check/preview/profile queries (seconds)
DQ_SAMPLE_LIMIT	10	Max bad rows captured per failing check
DQ_QUARANTINE_LIMIT	1000	Max rows quarantined per failing check
DQ_ALERT_WEBHOOK	unset	Default alert channel (scheme-routed: https/mailto/pagerduty/opsgenie)
DQ_ALERT_ROUTES	unset	JSON owner→channel map; values may be {fail: url, error: url}
DQ_SMTP_HOST	unset	SMTP host for mailto: routes
DQ_SMTP_PORT	587	SMTP port
DQ_SMTP_FROM	unset	From address for email alerts
DQ_SMTP_USER	unset	SMTP username (optional)
DQ_SMTP_PASSWORD	unset	SMTP password (optional)
DQ_OL_NAMESPACE	airflow-dq	Explicit OpenLineage namespace override (default: connection-derived)
DQ_CONTRACTS_GIT	false	Enable the git-based "propose" (PR) save flow
DQ_CONTRACTS_GIT_REMOTE	origin	Remote used by the propose flow
DQ_RESULTS_RETENTION_DAYS	0	Days to keep raw results/anomalies (0 = forever)

Upgrading from 0.1.x — breaking changes

• Freshness default is now mode: max_age — the check measures the age of max(col) (one aggregate, no per-row predicate) instead of counting every historical row older than the threshold. The 0.1 behavior is available as params: { mode: row_age }. unit: day|hour is now honored, and TIMESTAMP columns are supported.
• Quarantine no longer fires on tolerated passes — rows are quarantined only when the check's final status is fail or warn; a check whose failing rows are within tolerance (status pass) no longer writes quarantine rows.
• do_xcom_push defaults to off — check tasks no longer push their result dict (with bad-row samples) into XCom. Pass return_result=True to build_contract_checks() if you consumed it.
• Token-in-HTML removed — /dq/app/ no longer injects DQ_API_TOKEN into the page. The UI authenticates via the Airflow session cookie / auth-manager JWT; DQ_API_TOKEN still works as a Bearer header for machine clients. Any scraper that relied on reading the token from the HTML breaks (deliberately).
• ODCS v3.1 output shape — saved contracts now emit apiVersion: v3.1.0, a semver version string ("N.0.0", integer retained as the dqVersion custom property), and dataset/owner under customProperties. Readers accept both the old dialect and the new shape; unknown spec fields round-trip instead of being stripped.
• Infra errors are now status='error' — executor exceptions no longer crash record-only monitors; they are recorded (and alerted) as error, and only on_fail="fail" tasks re-raise.
• Contract versions are immutable — saving an existing version returns HTTP 409; the editor auto-targets max(existing)+1.

Production notes

• psycopg2: the wheel depends on psycopg2-binary for friction-free dev/demo installs. For production, follow the psycopg2 project's advice and install the source package instead (pip install psycopg2) so it links against your system libpq/libssl.
• Constraints: install with the Airflow constraints file matching your Airflow/Python version (CI does the same): pip install airflow-dq --constraint https://raw.githubusercontent.com/apache/airflow/constraints-3.1.0/constraints-3.12.txt
• Auth: Airflow 3 does not auto-protect plugin FastAPI routes. They are secured explicitly in plugin/auth.py: Airflow auth-manager JWT (header or session cookie) with DQ_API_TOKEN as a documented machine-client fallback; write routes additionally consult is_authorized_custom_view when available.
• UI hosting: the React app is served whole by the plugin's FastAPI app at /dq/app/ and embedded in the Airflow UI as a nav item via the stable external_views (iframe) API. Wheel installs serve the packaged bundle (airflow_dq/ui_dist); source checkouts serve ui/dist (build with make ui-build); DQ_UI_DIST overrides both.
• Experimental backends (Snowflake, BigQuery, Databricks, Trino): SQL generation is unit-tested per dialect, but no live-connection test coverage exists yet. Install the matching extra (pip install "airflow-dq[snowflake]" etc.) and treat them as preview quality.