diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 68fd3b5..3f6b764 100644 --- a/.github/workflows/ci.yml +++ b/.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: | diff --git a/README.md b/README.md index 01e6ac0..1d6b74d 100644 --- a/README.md +++ b/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) | diff --git a/dags/dqo_orders_checks.py b/dags/dqo_orders_checks.py new file mode 100644 index 0000000..01d1c49 --- /dev/null +++ b/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" + ), + ) diff --git a/docs/scheduling.md b/docs/scheduling.md new file mode 100644 index 0000000..faae358 --- /dev/null +++ b/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. diff --git a/requirements-airflow.txt b/requirements-airflow.txt new file mode 100644 index 0000000..a362384 --- /dev/null +++ b/requirements-airflow.txt @@ -0,0 +1 @@ +apache-airflow==2.10.4 diff --git a/tests/test_dags.py b/tests/test_dags.py new file mode 100644 index 0000000..b473dba --- /dev/null +++ b/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