Assembling the contract-driven quality gate

Description

This lesson writes run_full_gate() — the single entry point the earlier lesson's brief asked for. It isn't a long or complicated function: it's, literally, seven calls to already-existing functions, ordered, with its result normalized into a single shape. What makes it valuable isn't its complexity, it's its completeness: running it once guarantees this guide's seven checks ran, in the same order, on the same data, with nobody having to remember to call all seven separately.

Connection to the module. Lesson 2 wrote the brief. This lesson fulfills it: it builds run_full_gate(), tests it first on a toy DataFrame — to confirm the assembly works before touching the real incident — and leaves everything ready for lesson 4, where it runs for the first time on S04.

An analogy: the flight checklist, not seven loose inspections

Before a plane takes off, a pilot doesn't check each system separately, at different moments, trusting memory not to skip any. They follow a checklist — a fixed list, in a fixed order, where every item has a clear name and a binary result: checked or not checked. The checklist invents no new check — fuel, flaps, instruments already existed as individual systems before any list existed — what it adds is the guarantee that all of them get checked, always in the same order, and that every result gets recorded, not just remembered.

run_full_gate() is that checklist, applied to a Kiosko orders file. Every item — completeness, uniqueness, validity, consistency, accuracy, freshness, volume — is already a separately tested system, built in an earlier module. This lesson writes the list that walks through all of them, always in the same order, skipping none.

Worked example: run_full_gate(), piece by piece

Step 1 — the seven pieces, imported with no change

Everything that follows assumes you have, in the same working directory, modules 3 through 7's exact functions — with no modification to any of them — plus module 4's contract_to_pandera_schema() and DataContract/ColumnContract:

# kiosko_trust.py -- M3-M7's pieces, reused with no changes
import hashlib
from datetime import datetime
from typing import Literal

import pandera.polars as pa
import polars as pl
import yaml
from pydantic import BaseModel, Field

PIPELINE_RUN_AT = "2026-08-16T09:00:00"
REFERENCE_PRICES = {"P001": 0.55, "P002": 1.2, "P003": 0.75, "P004": 4.5}


# --- Module 4: the contract ---
class ColumnContract(BaseModel):
    name: str
    type: Literal["string", "float", "integer"]
    nullable: bool = True
    unique: bool = False
    minimum: float | None = None
    exclusive_minimum: float | None = None


class RowCountRange(BaseModel):
    min: int
    max: int


class SLAContract(BaseModel):
    freshness_hours: int
    row_count: RowCountRange


class DataContract(BaseModel):
    contract_version: str
    dataset: str
    owner: str
    description: str
    schema_: list[ColumnContract] = Field(alias="schema")
    sla: SLAContract
    on_violation: Literal["quarantine", "reject", "alert"]


TYPE_MAP = {"string": str, "float": float, "integer": int}


def load_contract(path: str) -> DataContract:
    with open(path) as f:
        raw = yaml.safe_load(f)
    return DataContract.model_validate(raw)


def contract_to_pandera_schema(contract: DataContract) -> pa.DataFrameSchema:
    columns = {}
    for col in contract.schema_:
        cast = int if col.type == "integer" else float
        checks = []
        if col.minimum is not None:
            checks.append(pa.Check.ge(cast(col.minimum)))
        if col.exclusive_minimum is not None:
            checks.append(pa.Check.gt(cast(col.exclusive_minimum)))
        columns[col.name] = pa.Column(
            TYPE_MAP[col.type], checks=checks, nullable=col.nullable, unique=col.unique
        )
    return pa.DataFrameSchema(columns)


# --- Module 3: consistency ---
def validate_referential_integrity(orders_df: pl.DataFrame, dim_product_df: pl.DataFrame) -> pl.DataFrame:
    return orders_df.join(dim_product_df, on="product_id", how="anti")


# --- Module 5: accuracy ---
def check_price_baseline(df: pl.DataFrame, reference_prices: dict[str, float], tolerance: float = 0.5) -> pl.DataFrame:
    return (
        df.with_columns(pl.col("product_id").replace_strict(reference_prices, default=None).alias("reference_price"))
        .filter(pl.col("unit_price").is_not_null() & pl.col("reference_price").is_not_null())
        .with_columns(((pl.col("unit_price") - pl.col("reference_price")).abs() / pl.col("reference_price")).alias("deviation"))
        .filter(pl.col("deviation") > tolerance)
    )


# --- Module 6: freshness, volume ---
def check_freshness(df: pl.DataFrame, run_at: str, sla_hours: int, timestamp_col: str = "order_ts") -> dict:
    latest_ts = df.select(pl.col(timestamp_col).max()).item()
    run_at_dt = datetime.fromisoformat(run_at)
    hours_since_latest = (run_at_dt - latest_ts).total_seconds() / 3600
    return {
        "check": "freshness", "latest_row_ts": str(latest_ts), "run_at": run_at,
        "sla_hours": sla_hours, "hours_since_latest": round(hours_since_latest, 2),
        "status": "PASS" if hours_since_latest <= sla_hours else "FAIL",
    }


def check_volume(df: pl.DataFrame, min_rows: int, max_rows: int) -> dict:
    row_count = df.height
    return {
        "check": "volume", "row_count": row_count, "min_rows": min_rows, "max_rows": max_rows,
        "status": "PASS" if min_rows <= row_count <= max_rows else "FAIL",
    }

None of this block is new. Every function has, exactly, the same signature and the same body you already saw, verified, in its origin module — contract_to_pandera_schema() from module 4, validate_referential_integrity() from module 3, check_price_baseline() from module 5, check_freshness()/check_volume() from module 6.

Step 2 — run_full_gate(), the complete checklist

def run_full_gate(
    df: pl.DataFrame,
    dim_product_df: pl.DataFrame,
    reference_prices: dict[str, float],
    schema: pa.DataFrameSchema,
    *,
    run_at: str,
    sla_hours: int = 24,
    min_rows: int = 5,
    max_rows: int = 20,
    tolerance: float = 0.5,
) -> list[dict]:
    """Runs this guide's seven checks, in order, on a single file.

    Returns a list of 7 results (one per check): completeness, uniqueness,
    validity (the three derived from the contract-generated schema), consistency,
    accuracy, freshness, volume. Every result has 'check', 'status' (PASS/FAIL)
    and 'detail'.
    """
    results: list[dict] = []

    # 1-3. completeness, uniqueness, validity -- generated from the contract (M4)
    try:
        schema.validate(df, lazy=True)
        schema_failures = pl.DataFrame(schema={"index": pl.UInt32, "column": pl.String, "check": pl.String, "failure_case": pl.String})
    except pa.errors.SchemaErrors as exc:
        schema_failures = exc.failure_cases

    for check_key, dimension in [("not_nullable", "completeness"), ("field_uniqueness", "uniqueness"), ("greater_than(0)", "validity")]:
        matches = schema_failures.filter(pl.col("check") == check_key)
        order_ids = sorted({df["order_id"][i] for i in matches["index"].to_list()})
        results.append({
            "check": dimension,
            "status": "FAIL" if matches.height > 0 else "PASS",
            "detail": f"{matches.height} physical rows: {order_ids}" if matches.height > 0 else "no rows",
        })

    # 4. consistency (M3)
    indexed = df.with_row_index("row_idx")
    orphans = validate_referential_integrity(indexed, dim_product_df)
    results.append({
        "check": "consistency",
        "status": "FAIL" if orphans.height > 0 else "PASS",
        "detail": f"{orphans.height} orphan rows: {orphans['order_id'].to_list()}" if orphans.height > 0 else "every product_id exists in dim_product",
    })

    # 5. accuracy (M5)
    anomalies = check_price_baseline(indexed, reference_prices, tolerance=tolerance)
    results.append({
        "check": "accuracy",
        "status": "FAIL" if anomalies.height > 0 else "PASS",
        "detail": f"{anomalies.height} rows outside the baseline: {anomalies['order_id'].to_list()}" if anomalies.height > 0 else "every price is within the baseline",
    })

    # 6. freshness (M6)
    freshness = check_freshness(df, run_at=run_at, sla_hours=sla_hours)
    results.append({
        "check": "freshness",
        "status": freshness["status"],
        "detail": f"{freshness['hours_since_latest']}h of {freshness['sla_hours']}h SLA",
    })

    # 7. volume (M6)
    volume = check_volume(df, min_rows=min_rows, max_rows=max_rows)
    results.append({
        "check": "volume",
        "status": volume["status"],
        "detail": f"{volume['row_count']} rows, range [{volume['min_rows']}, {volume['max_rows']}]",
    })

    return results


def gate_failure_count(gate_results: list[dict]) -> int:
    return sum(1 for r in gate_results if r["status"] == "FAIL")

Notice three design decisions, none accidental:

• The schema comes from the contract, not from a hand-written OrdersSchema. run_full_gate() receives schema as a parameter — an already-built pa.DataFrameSchema — instead of importing module 2's OrdersSchema class directly. This fulfills, literally, module 4's promise: the contract generates the test, and the complete system uses that generated version, not a parallel hand-written copy.
• The order is fixed, and matches the order in which this guide discovered each dimension. Completeness, uniqueness, validity (modules 1-2), consistency (module 3), accuracy (module 5), freshness and volume (module 6) — this entire guide's same narrative order, now as the system's real execution order.
• Every result is a dict with the exact same shape: check, status, detail. No row-level check (completeness, uniqueness, validity, consistency, accuracy) or file-level one (freshness, volume) has a different output format — that uniformity is what lets gate_failure_count() count all seven results with a single line, with no seven different if statements depending on check type.

Step 3 — the first run, on a toy DataFrame

Before touching S04's real incident, it's worth confirming the assembly works with a minimal, controlled example — the same discipline modules 3 and 5 already used before running their own functions on real data.

# toy_gate_test.py
import duckdb
import polars as pl
from kiosko_trust import (
    PIPELINE_RUN_AT, REFERENCE_PRICES, contract_to_pandera_schema, gate_failure_count,
    load_contract, run_full_gate,
)

con = duckdb.connect("kiosko.duckdb")
dim_product_df = con.sql("SELECT * FROM dim_product").pl()

contract = load_contract("orders_contract.yaml")
schema = contract_to_pandera_schema(contract)

toy_df = pl.DataFrame({
    "order_id": ["ORD-TOY-01", "ORD-TOY-02", "ORD-TOY-03"],
    "store_id": ["S01", "S01", "S02"],
    "product_id": ["P001", "P002", "P003"],
    "quantity": [2, 1, 3],
    "unit_price": [0.55, None, 0.75],
    "order_ts": ["2026-08-16T07:00:00", "2026-08-16T07:10:00", "2026-08-16T07:20:00"],
}).with_columns(pl.col("order_ts").str.to_datetime())

print("toy_df:")
print(toy_df)

toy_gate = run_full_gate(
    toy_df, dim_product_df, REFERENCE_PRICES, schema,
    run_at=PIPELINE_RUN_AT, sla_hours=contract.sla.freshness_hours,
    min_rows=1, max_rows=20,
)
print("\n=== run_full_gate() on toy_df ===")
for r in toy_gate:
    print(f"  [{r['status']}] {r['check']:<14} {r['detail']}")

print(f"\nFailures: {gate_failure_count(toy_gate)} of {len(toy_gate)} checks")

toy_df has three rows, deliberately simple: ORD-TOY-01 and ORD-TOY-03 are perfectly clean (exact REFERENCE_PRICES price, existing product, positive quantity, unique order_id); ORD-TOY-02 has unit_price=None — a single problem, in a single dimension, on purpose. dim_product gets read from kiosko.duckdb, the same four-product table modules 3 and 5 already used.

What to expect (verified by actually running python3 toy_gate_test.py, with kiosko.duckdb and orders_contract.yaml in the same folder, pandera==0.32.1):

toy_df:
shape: (3, 6)
┌────────────┬──────────┬────────────┬──────────┬────────────┬─────────────────────┐
│ order_id   ┆ store_id ┆ product_id ┆ quantity ┆ unit_price ┆ order_ts            │
│ ---        ┆ ---      ┆ ---        ┆ ---      ┆ ---        ┆ ---                 │
│ str        ┆ str      ┆ str        ┆ i64      ┆ f64        ┆ datetime[μs]        │
╞════════════╪══════════╪════════════╪══════════╪════════════╪═════════════════════╡
│ ORD-TOY-01 ┆ S01      ┆ P001       ┆ 2        ┆ 0.55       ┆ 2026-08-16 07:00:00 │
│ ORD-TOY-02 ┆ S01      ┆ P002       ┆ 1        ┆ null       ┆ 2026-08-16 07:10:00 │
│ ORD-TOY-03 ┆ S02      ┆ P003       ┆ 3        ┆ 0.75       ┆ 2026-08-16 07:20:00 │
└────────────┴──────────┴────────────┴──────────┴────────────┴─────────────────────┘

=== run_full_gate() on toy_df ===
  [FAIL] completeness   1 physical rows: ['ORD-TOY-02']
  [PASS] uniqueness     no rows
  [PASS] validity       no rows
  [PASS] consistency    every product_id exists in dim_product
  [PASS] accuracy       every price is within the baseline
  [PASS] freshness      1.67h of 24h SLA
  [PASS] volume         3 rows, range [1, 20]

Failures: 1 of 7 checks

Exactly what was expected: 1 failure, in the single dimension toy_df deliberately breaks, and the other six at PASS. freshness passes because toy_df's order_ts values are all from the same 2026-08-16, barely a few hours before PIPELINE_RUN_AT — a reference time deliberately close by, so this toy example doesn't get distracted by a freshness failure that isn't the point yet. This result confirms something important before moving on: the assembly works — the seven checks ran, in order, on the same data, and the failure count matches exactly what anyone looking at toy_df would expect at a glance.

Diagram: run_full_gate()'s complete flow

flowchart TD
    C["orders_contract.yaml"] -->|"load_contract() + M4"| D["DataContract"]
    D -->|"contract_to_pandera_schema()"| S["pa.DataFrameSchema"]

    F["orders file\n(df: pl.DataFrame)"] --> G["run_full_gate()"]
    S --> G
    P["dim_product_df"] --> G
    R["REFERENCE_PRICES"] --> G

    G --> R1["1-3. schema.validate()\ncompleteness/uniqueness/validity"]
    G --> R2["4. validate_referential_integrity()\nconsistency"]
    G --> R3["5. check_price_baseline()\naccuracy"]
    G --> R4["6. check_freshness()\nfreshness"]
    G --> R5["7. check_volume()\nvolume"]

    R1 --> OUT["list of 7 dict\ncheck/status/detail"]
    R2 --> OUT
    R3 --> OUT
    R4 --> OUT
    R5 --> OUT

Common mistakes

Accidentally running schema.validate() twice, once in the try and again outside it. What happens: someone, reading the completeness/uniqueness/validity block, calls schema.validate(df, lazy=True) again later in the code, thinking they need the "non-lazy" result for something. Why it happens: lazy=True isn't intuitive the first time you use it — without it, .validate() raises the exception on the first error it finds, instead of accumulating them all in SchemaErrors.failure_cases. How to spot it: if your code calls .validate() on the same df and the same schema more than once, you're repeating work — Pandera already evaluated all three rules (nullable, unique, the range Checks) in a single pass. How to fix it: a single call, with lazy=True, inside the try/except pa.errors.SchemaErrors block — exactly how run_full_gate() does it, the same pattern modules 2 and 4 already used.

Forgetting indexed = df.with_row_index("row_idx") needs to be passed to validate_referential_integrity() and check_price_baseline(), not the un-indexed df. What happens: someone copies run_full_gate() but calls validate_referential_integrity(df, dim_product_df) instead of validate_referential_integrity(indexed, dim_product_df), and the result still works — but it loses the row_idx column other parts of the system (lesson 4, with build_failure_report()) are going to need later. Why it happens: validate_referential_integrity() doesn't need row_idx to work on its own — the anti-join works the same with or without that column — so the mistake doesn't show up right away. How to spot it: if your orphans or anomalies have no row_idx column, any code depending on it to quarantine rows (lesson 4) is going to fail later, with a missing-column message. How to fix it: index the DataFrame once, before passing it to any function that needs to track specific physical rows — exactly the pattern build_failure_report() already followed in module 7.

Exercises

Exercise 1 — Run toy_gate_test.py yourself, and then break a second dimension on purpose. With kiosko.duckdb and orders_contract.yaml in your folder, run this lesson's script and confirm you see 1 failure. Then, change ORD-TOY-03's quantity to -2, run again, and confirm how many failures you see now.

Exercise 2 — Extend run_full_gate() to also receive an optional verbose: bool = False parameter that prints each check as it runs, instead of only returning the list at the end. Modify the function so that, if verbose=True, it prints f"[{status}] {check}: {detail}" immediately after calculating each result.

def run_full_gate(
    df: pl.DataFrame, dim_product_df: pl.DataFrame, reference_prices: dict[str, float],
    schema: pa.DataFrameSchema, *, run_at: str, sla_hours: int = 24, min_rows: int = 5,
    max_rows: int = 20, tolerance: float = 0.5, verbose: bool = False,
) -> list[dict]:
    results: list[dict] = []

    def _add(result: dict) -> None:
        results.append(result)
        if verbose:
            print(f"  [{result['status']}] {result['check']}: {result['detail']}")

    # ... (the same body as before, replacing each results.append(...) with _add(...))
    return results

Exercise 3 — Argue why run_full_gate() receives schema as a parameter, instead of building it itself from a file path (contract_path: str). In 2-3 sentences, considering how this function gets tested in today's lesson (on toy_df, with no S04-specific contract) and how it's going to be used in lesson 4 (on real data), argue in favor of the current design.

Summary and next step

In this lesson you built run_full_gate(), the single entry point lesson 2's brief asked for: seven checks, in a fixed order, with a structured, uniform result — check/status/detail — for each one. You confirmed, on a three-row toy DataFrame with a single deliberate problem, that the assembly works exactly as expected: one failure, in the correct dimension, and the other six at PASS.

Before moving on you should be able to: name, in order, the seven checks run_full_gate() runs; and explain why it receives an already-built schema instead of a contract file path.

Lesson 4 runs this exact same system, with no change at all, on the real incident: orders_2026-08-14.csv, the S04 file that opened this guide in module 1. That's where run_full_gate() stops being a three-row exercise and becomes the definitive proof of everything this guide built.

Resources

• Pandera — official documentation (DataFrameSchema, lazy, SchemaErrors.failure_cases), the technical foundation of the first three checks. pandera.readthedocs.io. In English.
• Module 4, lesson 5, of this same guide — the exact source of contract_to_pandera_schema(), reused with no changes in this lesson. src/guides/data-reliability-and-governance-guide/workbook/module-04-data-contracts-as-versioned-artifacts/en/05-from-contract-to-pandera-schema.md. In English.
• Module 6, lessons 4-5, of this same guide — the exact source of check_freshness() and check_volume(). src/guides/data-reliability-and-governance-guide/workbook/module-06-freshness-volume-and-lineage/en/. In English.
• This guide's DESIGN. src/guides/data-reliability-and-governance-guide/DISENO.md. In Spanish.