From d003a5fa1f07fc8937d6cc402e18dc764bf79291 Mon Sep 17 00:00:00 2001 From: br413 Date: Tue, 14 Jul 2026 18:20:04 -0600 Subject: [PATCH] feat: extend contract checks DAG with customers task --- README.md | 6 ++-- ...rders_checks.py => dqo_contract_checks.py} | 36 +++++++++++++------ docs/scheduling.md | 4 +-- tests/test_dags.py | 6 ++-- 4 files changed, 33 insertions(+), 19 deletions(-) rename dags/{dqo_orders_checks.py => dqo_contract_checks.py} (58%) diff --git a/README.md b/README.md index 7fbb70a..a0cba24 100644 --- a/README.md +++ b/README.md @@ -38,7 +38,7 @@ Scheduling options are documented in [`docs/scheduling.md`](docs/scheduling.md). - [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 -- [x] Airflow DAG `dqo_orders_contract_checks` for scheduled contract runs +- [x] Airflow DAG `dqo_contract_checks` for scheduled contract runs - [ ] Webhook alert integration tests against mock server ## Technology @@ -99,7 +99,7 @@ python -m src.dqo.cli history --contract orders │ ├── workflows/ci.yml │ └── pull_request_template.md ├── dags/ -│ └── dqo_orders_checks.py +│ └── dqo_contract_checks.py ├── contracts/ │ ├── orders.yml │ └── customers.yml @@ -135,7 +135,7 @@ Coverage includes contract loading, each check type, end-to-end runs, history pe | Concern | Approach | |---------|----------| -| Scheduling | Airflow DAG `@daily` (`dqo_orders_contract_checks`) | +| Scheduling | Airflow DAG `@daily` (`dqo_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_contract_checks.py similarity index 58% rename from dags/dqo_orders_checks.py rename to dags/dqo_contract_checks.py index 01d1c49..ba33944 100644 --- a/dags/dqo_orders_checks.py +++ b/dags/dqo_contract_checks.py @@ -21,18 +21,26 @@ HISTORY_DB = os.environ.get("DQO_DATABASE_URL", "sqlite:///.dqo/history.db") ALERT_FILE = os.environ.get("DQO_ALERT_FILE", ".dqo/alerts.jsonl") +CHECK_COMMAND = ( + "python -m src.dqo.cli run " + "--references data/samples " + f"--history-db {HISTORY_DB} " + f"--alert-file {ALERT_FILE} " + "--no-console-alerts" +) + with DAG( - dag_id="dqo_orders_contract_checks", + dag_id="dqo_contract_checks", default_args=DEFAULT_ARGS, - description="Run orders data contract checks and persist history", + description="Run orders and customers contract checks", schedule="@daily", start_date=datetime(2026, 7, 1), catchup=False, tags=["data-quality", "contracts", "portfolio"], doc_md=""" - ## dqo_orders_contract_checks + ## dqo_contract_checks - 1. Execute YAML contract checks against sample orders data + 1. Execute orders and customers YAML contract checks 2. Persist run history and route alerts to JSONL Set `DQO_PROJECT_ROOT` to the repository root when deploying. @@ -41,13 +49,19 @@ run_orders_checks = BashOperator( task_id="run_orders_checks", bash_command=( - f"cd {PROJECT_ROOT} && " - "python -m src.dqo.cli run " + f"cd {PROJECT_ROOT} && {CHECK_COMMAND} " "--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" + "--data data/samples/orders.csv" + ), + ) + + run_customers_checks = BashOperator( + task_id="run_customers_checks", + bash_command=( + f"cd {PROJECT_ROOT} && {CHECK_COMMAND} " + "--contract contracts/customers.yml " + "--data data/samples/customers.csv" ), ) + + run_orders_checks >> run_customers_checks diff --git a/docs/scheduling.md b/docs/scheduling.md index faae358..3affadf 100644 --- a/docs/scheduling.md +++ b/docs/scheduling.md @@ -2,11 +2,11 @@ ## Airflow -The `dags/dqo_orders_contract_checks.py` DAG runs the orders contract daily: +The `dags/dqo_contract_checks.py` DAG runs orders and customers contracts daily: ```bash export DQO_PROJECT_ROOT=/path/to/data-quality-observability -airflow dags test dqo_orders_contract_checks 2026-07-14 +airflow dags test dqo_contract_checks 2026-07-14 ``` Environment variables: diff --git a/tests/test_dags.py b/tests/test_dags.py index b473dba..2632894 100644 --- a/tests/test_dags.py +++ b/tests/test_dags.py @@ -23,16 +23,16 @@ def test_dag_loads_without_import_errors() -> None: 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") + dag = dag_bag.get_dag("dqo_contract_checks") assert dag is not None - assert len(dag.tasks) == 1 + assert len(dag.tasks) == 2 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") + dag = dag_bag.get_dag("dqo_contract_checks") assert dag is not None assert dag.schedule_interval == "@daily" assert "data-quality" in dag.tags