Metadata-Version: 2.4
Name: airflow-dq
Version: 0.2.1
Summary: Airflow-3-native data-quality & data-contract plugin with an integrated quality dashboard
Project-URL: Homepage, https://github.com/fbnschneider/airflow-dq
Project-URL: Repository, https://github.com/fbnschneider/airflow-dq
Project-URL: Changelog, https://github.com/fbnschneider/airflow-dq/blob/main/CHANGELOG.md
Project-URL: Issues, https://github.com/fbnschneider/airflow-dq/issues
Author-email: Fabian Schneider <fabian.schneider.dev@gmail.com>
License: Apache-2.0
License-File: LICENSE
License-File: NOTICE
Keywords: airflow,data-contracts,data-quality,odcs,openlineage
Classifier: Development Status :: 4 - Beta
Classifier: Framework :: Apache Airflow
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: Apache Software License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Database
Requires-Python: >=3.10
Requires-Dist: apache-airflow>=3.1.0
Requires-Dist: duckdb>=0.10
Requires-Dist: fastapi>=0.110
Requires-Dist: psycopg2-binary>=2.9
Requires-Dist: pydantic>=2.6
Requires-Dist: pyyaml>=6.0
Requires-Dist: sqlalchemy>=1.4.36
Provides-Extra: bigquery
Requires-Dist: sqlalchemy-bigquery>=1.9; extra == 'bigquery'
Provides-Extra: databricks
Requires-Dist: databricks-sql-connector>=3.0; extra == 'databricks'
Requires-Dist: databricks-sqlalchemy>=2.0; extra == 'databricks'
Provides-Extra: dev
Requires-Dist: jsonschema>=4.21; extra == 'dev'
Requires-Dist: mypy>=1.9; extra == 'dev'
Requires-Dist: pytest-cov>=4.1; extra == 'dev'
Requires-Dist: pytest>=8.0; extra == 'dev'
Requires-Dist: ruff>=0.4; extra == 'dev'
Requires-Dist: types-pyyaml; extra == 'dev'
Provides-Extra: lineage
Requires-Dist: apache-airflow-providers-openlineage>=2.0; extra == 'lineage'
Provides-Extra: snowflake
Requires-Dist: snowflake-sqlalchemy>=1.5; extra == 'snowflake'
Provides-Extra: trino
Requires-Dist: trino[sqlalchemy]>=0.328; extra == 'trino'
Description-Content-Type: text/markdown

# airflow-dq

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

You author [ODCS](https://github.com/bitol-io/open-data-contract-standard) 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](LICENSE) · [Changelog](CHANGELOG.md) · [Contributing](CONTRIBUTING.md) ·
[Security policy](SECURITY.md). 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](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

```mermaid
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

```bash
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](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

```python
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](src/airflow_dq/config.py)).
See [.env.example](.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`](src/airflow_dq/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.
