feat: add Airflow DAG for scheduled contract checks

.github/workflows/ci.yml

@@ -43,9 +43,17 @@ jobs:
         run: |
           python -m pip install --upgrade pip
           pip install -r requirements.txt
+          pip install -r requirements-airflow.txt
+
+      - name: Initialize Airflow metadata database
+        run: airflow db migrate
+        env:
+          AIRFLOW_HOME: /tmp/airflow
 
       - name: Run tests
         run: python -m pytest -q
+        env:
+          AIRFLOW_HOME: /tmp/airflow
 
       - name: Smoke test CLI
         run: |

README.md

@@ -27,6 +27,8 @@ On-call triage runbook
 
 See [`docs/architecture.md`](docs/architecture.md) for component boundaries and failure modes.
 
+Scheduling options are documented in [`docs/scheduling.md`](docs/scheduling.md).
+
 ## Current capabilities
 
 - [x] YAML data contracts with column rules and foreign keys
@@ -36,7 +38,8 @@ See [`docs/architecture.md`](docs/architecture.md) for component boundaries and
 - [x] Alert routing to console, JSONL file, and optional webhook
 - [x] Failure triage runbook in [`docs/operations.md`](docs/operations.md)
 - [x] Unit and integration tests with CI
-- [ ] Scheduled orchestration (Airflow / cron integration planned)
+- [x] Airflow DAG `dqo_orders_contract_checks` for scheduled contract runs
+- [ ] Webhook alert integration tests against mock server
 
 ## Technology
 
@@ -95,6 +98,8 @@ python -m src.dqo.cli history --contract orders
 │   ├── ISSUE_TEMPLATE/
 │   ├── workflows/ci.yml
 │   └── pull_request_template.md
+├── dags/
+│   └── dqo_orders_checks.py
 ├── contracts/
 │   └── orders.yml
 ├── data/samples/
@@ -129,7 +134,7 @@ Coverage includes contract loading, each check type, end-to-end runs, history pe
 
 | Concern | Approach |
 |---------|----------|
-| Scheduling | Manual CLI today; orchestrator hook planned |
+| Scheduling | Airflow DAG `@daily` (`dqo_orders_contract_checks`) |
 | Monitoring | Check history + alert JSONL |
 | Retries | Re-run after upstream fix; history preserves prior failures |
 | Triage | [`docs/operations.md`](docs/operations.md) |

dags/dqo_orders_checks.py

@@ -0,0 +1,53 @@
+"""Airflow DAG for scheduled data-quality contract checks."""
+
+from __future__ import annotations
+
+import os
+from datetime import datetime, timedelta
+
+from airflow import DAG
+from airflow.operators.bash import BashOperator
+
+DEFAULT_ARGS = {
+    "owner": "br413",
+    "depends_on_past": False,
+    "email_on_failure": False,
+    "email_on_retry": False,
+    "retries": 1,
+    "retry_delay": timedelta(minutes=5),
+}
+
+PROJECT_ROOT = os.environ.get("DQO_PROJECT_ROOT", os.getcwd())
+HISTORY_DB = os.environ.get("DQO_DATABASE_URL", "sqlite:///.dqo/history.db")
+ALERT_FILE = os.environ.get("DQO_ALERT_FILE", ".dqo/alerts.jsonl")
+
+with DAG(
+    dag_id="dqo_orders_contract_checks",
+    default_args=DEFAULT_ARGS,
+    description="Run orders data contract checks and persist history",
+    schedule="@daily",
+    start_date=datetime(2026, 7, 1),
+    catchup=False,
+    tags=["data-quality", "contracts", "portfolio"],
+    doc_md="""
+    ## dqo_orders_contract_checks
+
+    1. Execute YAML contract checks against sample orders data
+    2. Persist run history and route alerts to JSONL
+
+    Set `DQO_PROJECT_ROOT` to the repository root when deploying.
+    """,
+) as dag:
+    run_orders_checks = BashOperator(
+        task_id="run_orders_checks",
+        bash_command=(
+            f"cd {PROJECT_ROOT} && "
+            "python -m src.dqo.cli run "
+            "--contract contracts/orders.yml "
+            "--data data/samples/orders.csv "
+            "--references data/samples "
+            f"--history-db {HISTORY_DB} "
+            f"--alert-file {ALERT_FILE} "
+            "--no-console-alerts"
+        ),
+    )

docs/scheduling.md

@@ -0,0 +1,31 @@
+# Scheduling data-quality checks
+
+## Airflow
+
+The `dags/dqo_orders_contract_checks.py` DAG runs the orders contract daily:
+
+```bash
+export DQO_PROJECT_ROOT=/path/to/data-quality-observability
+airflow dags test dqo_orders_contract_checks 2026-07-14
+```
+
+Environment variables:
+
+| Variable | Default | Purpose |
+|----------|---------|---------|
+| `DQO_PROJECT_ROOT` | current directory | Repository root for CLI invocation |
+| `DQO_DATABASE_URL` | `sqlite:///.dqo/history.db` | History store connection |
+| `DQO_ALERT_FILE` | `.dqo/alerts.jsonl` | Alert JSONL destination |
+
+## Manual / cron
+
+For lightweight schedules without Airflow:
+
+```bash
+python -m src.dqo.cli run \
+  --contract contracts/orders.yml \
+  --data data/samples/orders.csv \
+  --references data/samples
+```
+
+Pair with cron or a systemd timer in environments where Airflow is not deployed.

requirements-airflow.txt

@@ -0,0 +1 @@
+apache-airflow==2.10.4

tests/test_dags.py

@@ -0,0 +1,38 @@
+"""Validate Airflow DAG integrity."""
+
+from __future__ import annotations
+
+import os
+import sys
+from pathlib import Path
+
+import pytest
+
+pytest.importorskip("airflow")
+
+PROJECT_ROOT = Path(__file__).resolve().parents[1]
+DAGS_FOLDER = str(PROJECT_ROOT / "dags")
+
+sys.path.insert(0, DAGS_FOLDER)
+os.environ.setdefault("AIRFLOW_HOME", str(PROJECT_ROOT / ".airflow"))
+
+
+def test_dag_loads_without_import_errors() -> None:
+    from airflow.models import DagBag
+
+    dag_bag = DagBag(dag_folder=DAGS_FOLDER, include_examples=False)
+    assert dag_bag.import_errors == {}, f"DAG import errors: {dag_bag.import_errors}"
+
+    dag = dag_bag.get_dag("dqo_orders_contract_checks")
+    assert dag is not None
+    assert len(dag.tasks) == 1
+
+
+def test_dag_schedule_and_tags() -> None:
+    from airflow.models import DagBag
+
+    dag_bag = DagBag(dag_folder=DAGS_FOLDER, include_examples=False)
+    dag = dag_bag.get_dag("dqo_orders_contract_checks")
+    assert dag is not None
+    assert dag.schedule_interval == "@daily"
+    assert "data-quality" in dag.tags