Week 11: registry CLI resolution and Airflow DAG improvements

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
 ```
 

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,40 +29,55 @@
     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,
     tags=["data-quality", "contracts", "portfolio"],
     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

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)

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

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,26 +35,33 @@ 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]]] = {}
     if reference_dir is not None:
         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,

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

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")