diff --git a/README.md b/README.md index 4f939ac..03b12a7 100644 --- a/README.md +++ b/README.md @@ -50,7 +50,7 @@ Scheduling options are documented in [`docs/scheduling.md`](docs/scheduling.md). - [x] Airflow DAG `dqo_contract_checks` for scheduled contract runs - [x] Webhook alert integration tests against mock server - [x] Contract registry catalog (`contracts/registry.yml`) — [ADR 0002](docs/adr/0002-schema-registry-and-contract-versioning.md) -- [ ] CLI resolves `--contract orders` via registry (phase 2) +- [x] CLI resolves `--contract orders` via registry (phase 2) - [ ] Run history stores `contract_version` metadata (phase 3) ## Technology stack @@ -79,7 +79,7 @@ Windows: .\.venv\Scripts\Activate.ps1 pip install -r requirements.txt pytest -python -m src.dqo.cli run --contract contracts/orders.yml --data data/samples/orders.csv --references data/samples +python -m src.dqo.cli run --contract orders --data data/samples/orders.csv --references data/samples ``` Linux/macOS: @@ -88,7 +88,7 @@ Linux/macOS: source .venv/bin/activate pip install -r requirements.txt pytest -python -m src.dqo.cli run --contract contracts/orders.yml --data data/samples/orders.csv --references data/samples +python -m src.dqo.cli run --contract orders --data data/samples/orders.csv --references data/samples ``` Run the demo script (Windows): @@ -109,7 +109,7 @@ After landing data through the companion ingestion pipeline (including quarantin ```bash # From production-data-pipeline: ingest sample events, then return here -python -m src.dqo.cli run --contract contracts/orders.yml --data data/samples/orders.csv --references data/samples +python -m src.dqo.cli run --contract orders --data data/samples/orders.csv --references data/samples python -m src.dqo.cli run --contract contracts/customers.yml --data data/samples/customers.csv --references data/samples ``` diff --git a/dags/dqo_contract_checks.py b/dags/dqo_contract_checks.py index ba33944..c455202 100644 --- a/dags/dqo_contract_checks.py +++ b/dags/dqo_contract_checks.py @@ -20,6 +20,7 @@ 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") +WEBHOOK_URL = os.environ.get("DQO_WEBHOOK_URL", "") CHECK_COMMAND = ( "python -m src.dqo.cli run " @@ -28,11 +29,27 @@ f"--alert-file {ALERT_FILE} " "--no-console-alerts" ) +if WEBHOOK_URL: + CHECK_COMMAND += f' --webhook-url "{WEBHOOK_URL}"' + + +def _contract_task(task_id: str, contract_name: str, dataset: str) -> BashOperator: + return BashOperator( + task_id=task_id, + bash_command=( + f"cd {PROJECT_ROOT} && " + f'echo "Running contract {contract_name} via registry" && ' + f"{CHECK_COMMAND} " + f"--contract {contract_name} " + f"--data data/samples/{dataset}" + ), + ) + with DAG( dag_id="dqo_contract_checks", default_args=DEFAULT_ARGS, - description="Run orders and customers contract checks", + description="Run orders and customers contract checks via registry", schedule="@daily", start_date=datetime(2026, 7, 1), catchup=False, @@ -40,28 +57,27 @@ doc_md=""" ## dqo_contract_checks - 1. Execute orders and customers YAML contract checks - 2. Persist run history and route alerts to JSONL + Scheduled dataset contract checks resolved through `contracts/registry.yml`. + + 1. **run_orders_checks** — `orders@1.0` against sample CSV + 2. **run_customers_checks** — `customers@1.0` against sample CSV + + Each task persists run history and appends alert JSONL. Set optional `DQO_WEBHOOK_URL` + for webhook routing on failures (same pattern as `production-data-pipeline`). + + Environment: - Set `DQO_PROJECT_ROOT` to the repository root when deploying. + | Variable | Purpose | + |----------|---------| + | `DQO_PROJECT_ROOT` | Repository root on the Airflow worker | + | `DQO_DATABASE_URL` | History store (SQLite default) | + | `DQO_ALERT_FILE` | JSONL alert output path | + | `DQO_WEBHOOK_URL` | Optional webhook for contract failures | + + See [ADR 0002](docs/adr/0002-schema-registry-and-contract-versioning.md) for registry design. """, ) as dag: - run_orders_checks = BashOperator( - task_id="run_orders_checks", - bash_command=( - f"cd {PROJECT_ROOT} && {CHECK_COMMAND} " - "--contract contracts/orders.yml " - "--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 = _contract_task("run_orders_checks", "orders", "orders.csv") + run_customers_checks = _contract_task("run_customers_checks", "customers", "customers.csv") run_orders_checks >> run_customers_checks diff --git a/src/dqo/cli.py b/src/dqo/cli.py index a61c2b2..0acccfb 100644 --- a/src/dqo/cli.py +++ b/src/dqo/cli.py @@ -25,9 +25,15 @@ def build_parser() -> argparse.ArgumentParser: subparsers = parser.add_subparsers(dest="command", required=True) run_parser = subparsers.add_parser("run", help="Execute checks for a contract") - run_parser.add_argument("--contract", required=True, type=Path) + run_parser.add_argument( + "--contract", + required=True, + help="Registry contract name (e.g. orders) or path to contracts/*.yml", + ) run_parser.add_argument("--data", required=True, type=Path) run_parser.add_argument("--references", type=Path, default=None) + run_parser.add_argument("--registry", type=Path, default=Path("contracts/registry.yml")) + run_parser.add_argument("--contracts-dir", type=Path, default=Path("contracts")) run_parser.add_argument("--history-db", type=str, default=None) run_parser.add_argument("--alert-file", type=Path, default=Path(".dqo/alerts.jsonl")) run_parser.add_argument("--webhook-url", type=str, default=None) @@ -57,6 +63,8 @@ def main(argv: list[str] | None = None) -> int: args.data, reference_dir=args.references, now=args.reference_time, + registry_path=args.registry, + contracts_dir=args.contracts_dir, ) store = HistoryStore(database_url=args.history_db) diff --git a/src/dqo/registry.py b/src/dqo/registry.py new file mode 100644 index 0000000..d82f37d --- /dev/null +++ b/src/dqo/registry.py @@ -0,0 +1,58 @@ +"""Resolve contract names via contracts/registry.yml.""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any + +import yaml + + +def load_registry(registry_path: Path) -> dict[str, Any]: + if not registry_path.is_file(): + raise FileNotFoundError(f"Contract registry not found: {registry_path}") + + payload = yaml.safe_load(registry_path.read_text(encoding="utf-8")) + if not isinstance(payload, dict): + raise ValueError("registry file must be a mapping") + + contracts = payload.get("contracts") + if not isinstance(contracts, dict): + raise ValueError("registry contracts section must be a mapping") + + return contracts + + +def resolve_contract_path( + name_or_path: str | Path, + *, + registry_path: Path = Path("contracts/registry.yml"), + contracts_dir: Path = Path("contracts"), +) -> Path: + candidate = Path(name_or_path) + + if candidate.suffix == ".yml": + if candidate.is_file(): + return candidate + nested = contracts_dir / candidate.name + if nested.is_file(): + return nested + + contract_key = str(name_or_path) + registry = load_registry(registry_path) + entry = registry.get(contract_key) + if not isinstance(entry, dict): + raise ValueError( + f"Unknown contract '{contract_key}'. " + f"Use a registry name or path to a .yml file under {contracts_dir}/" + ) + + relative_path = entry.get("path") + if not isinstance(relative_path, str): + raise ValueError(f"Registry entry for '{contract_key}' is missing a path") + + resolved = contracts_dir / relative_path + if not resolved.is_file(): + raise FileNotFoundError(f"Contract file for '{contract_key}' not found: {resolved}") + + return resolved diff --git a/src/dqo/runner.py b/src/dqo/runner.py index 3fdade8..c49d06c 100644 --- a/src/dqo/runner.py +++ b/src/dqo/runner.py @@ -12,6 +12,7 @@ from src.dqo.checks.schema import validate_schema from src.dqo.checks.uniqueness import validate_uniqueness from src.dqo.contracts import load_contract +from src.dqo.registry import resolve_contract_path from src.dqo.dataset import load_csv from src.dqo.models import CheckResult, DataContract, RunSummary @@ -34,14 +35,21 @@ def run_checks( def run_contract_file( - contract_path: Path, + contract: Path | str, dataset_path: Path, *, reference_dir: Path | None = None, now: datetime | None = None, + registry_path: Path = Path("contracts/registry.yml"), + contracts_dir: Path = Path("contracts"), ) -> RunSummary: started_at = datetime.now(timezone.utc) - contract = load_contract(contract_path) + contract_path = resolve_contract_path( + contract, + registry_path=registry_path, + contracts_dir=contracts_dir, + ) + loaded = load_contract(contract_path) rows = load_csv(dataset_path) references: dict[str, list[dict[str, str]]] = {} @@ -49,11 +57,11 @@ def run_contract_file( for csv_path in sorted(reference_dir.glob("*.csv")): references[csv_path.stem] = load_csv(csv_path) - results = run_checks(contract, rows, reference_tables=references, now=now) + results = run_checks(loaded, rows, reference_tables=references, now=now) finished_at = datetime.now(timezone.utc) return RunSummary( - contract_name=contract.name, + contract_name=loaded.name, run_id=str(uuid.uuid4()), started_at=started_at, finished_at=finished_at, diff --git a/tests/test_cli.py b/tests/test_cli.py index 5a3f788..51569c9 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -22,3 +22,24 @@ def test_cli_run_passes_with_reference_time(tmp_path: Path) -> None: ) assert exit_code == 0 + + +def test_cli_run_resolves_registry_contract_name(tmp_path: Path) -> None: + exit_code = main( + [ + "run", + "--contract", + "orders", + "--data", + "data/samples/orders.csv", + "--references", + "data/samples", + "--reference-time", + "2026-07-14T12:00:00Z", + "--no-console-alerts", + "--history-db", + f"sqlite:///{tmp_path / 'history.db'}", + ] + ) + + assert exit_code == 0 diff --git a/tests/test_registry.py b/tests/test_registry.py new file mode 100644 index 0000000..5ff78ff --- /dev/null +++ b/tests/test_registry.py @@ -0,0 +1,26 @@ +from pathlib import Path + +import pytest + +from src.dqo.registry import load_registry, resolve_contract_path + + +def test_load_registry_lists_orders_and_customers() -> None: + registry = load_registry(Path("contracts/registry.yml")) + assert set(registry) == {"orders", "customers"} + assert registry["orders"]["current"] == "1.0" + + +def test_resolve_contract_by_registry_name() -> None: + path = resolve_contract_path("orders") + assert path == Path("contracts/orders.yml") + + +def test_resolve_contract_by_relative_yml_path() -> None: + path = resolve_contract_path("contracts/orders.yml") + assert path == Path("contracts/orders.yml") + + +def test_resolve_unknown_contract_raises() -> None: + with pytest.raises(ValueError, match="Unknown contract 'missing'"): + resolve_contract_path("missing")