Skip to content

feat: extend contract checks DAG with customers task #5

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
br413 merged 1 commit into from
Jul 15, 2026
Merged

feat: extend contract checks DAG with customers task #5

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 3 additions & 3 deletions README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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) |
Expand Down
36 changes: 25 additions & 11 deletions dags/dqo_orders_checks.py → dags/dqo_contract_checks.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand All @@ -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

@cursor cursor Bot Jul 15, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Orders failure skips customers checks

Medium Severity

run_orders_checks is wired upstream of run_customers_checks, and the CLI exits with a non-zero code when contract checks fail. Airflow then skips the customers task, so a failing orders run leaves the customers contract unvalidated for that schedule despite independent data and contracts.

Fix in Cursor Fix in Web

Reviewed by Cursor Bugbot for commit d003a5f. Configure here.

4 changes: 2 additions & 2 deletions docs/scheduling.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down
6 changes: 3 additions & 3 deletions tests/test_dags.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Loading