Running the full gate against S04's incident

Description

Lesson 3 confirmed run_full_gate() works, on three toy rows with a single deliberate problem. This lesson runs it on the real file: orders_2026-08-14.csv, the same twelve S04 lines module 1 diagnosed by hand, module 2 started catching with Pandera, module 3 completed with referential integrity, module 4 formalized as a contract, module 5 closed with accuracy, and module 6 finished diagnosing with freshness and volume. Seven modules of work, condensed into a single call to a single function — and the result, 6 failures out of 7 checks, is the final confirmation that the integrated system sees exactly what the seven earlier modules saw, separately.

Connection to the module. This lesson discovers no new problem — every one of the six failures you're going to see, you already know, with evidence, from an earlier module. What's new is the shape: a single report, generated by a single function, also honoring the on_violation: quarantine policy module 4's contract declared from day one.

An analogy: the same audit, now with a single signed report

This guide's modules 1 through 7 are like seven auditors who each reviewed, on their own, a different area of the same company — accounting, inventory, regulatory compliance —, each delivering their own report on different dates. They all agree, in their individual findings, on the same six problems. What was missing was the consolidated report: a single document, signed once, gathering the seven auditors' six findings into a single table, with a clear final verdict. This lesson is that consolidated report, applied to S04's incident.

The material you need

You need, in a new working directory:

module_8_full_gate/
├── kiosko.duckdb              (orders_s04, dim_product -- already built in M2-M3)
├── orders_contract.yaml       (module 4, lesson 3)
└── kiosko_trust.py            (this module's lesson 3)

If your kiosko.duckdb doesn't have orders_s04 or dim_product yet, repeat module 2's lesson 4, step 2 and module 3's lesson 4, step 1 — this lesson doesn't re-explain those steps, it assumes you've already done them.

Worked example: the real run, plus quarantine and alert

Step 1 — build_failure_report(), adapted to receive the contract's schema

quarantine() needs to know, row by row, which ones have some problem — the same work build_failure_report() already did in module 7, with a single change: instead of importing the hand-written OrdersSchema class, it receives the schema already generated from the contract, exactly the same one run_full_gate() uses.

# add to kiosko_trust.py
CHECK_TO_DIMENSION = {
    "not_nullable": "completeness",
    "field_uniqueness": "uniqueness",
    "greater_than(0)": "validity",
}


def build_failure_report(
    df: pl.DataFrame,
    dim_product_df: pl.DataFrame,
    reference_prices: dict[str, float],
    schema: pa.DataFrameSchema,
) -> pl.DataFrame:
    """Same as module 7's build_failure_report(), with a single change: it receives
    the schema GENERATED from the contract (M4) instead of the hand-written
    OrdersSchema class -- the same union of completeness/uniqueness/validity/
    consistency/accuracy into a single table of failures per physical row."""
    rows: list[dict] = []
    try:
        schema.validate(df, lazy=True)
    except pa.errors.SchemaErrors as exc:
        for r in exc.failure_cases.iter_rows(named=True):
            rows.append({"row_idx": r["index"], "order_id": df["order_id"][r["index"]],
                         "dimension": CHECK_TO_DIMENSION[r["check"]], "detail": f"{r['column']}={r['failure_case']}"})
    indexed = df.with_row_index("row_idx")
    for r in validate_referential_integrity(indexed, dim_product_df).iter_rows(named=True):
        rows.append({"row_idx": r["row_idx"], "order_id": r["order_id"], "dimension": "consistency",
                     "detail": f"product_id={r['product_id']} does not exist in dim_product"})
    for r in check_price_baseline(indexed, reference_prices, tolerance=0.5).iter_rows(named=True):
        rows.append({"row_idx": r["row_idx"], "order_id": r["order_id"], "dimension": "accuracy",
                     "detail": f"unit_price={r['unit_price']} is {round(r['deviation'], 1)}x away from the reference price ({r['reference_price']})"})
    if not rows:
        return pl.DataFrame(schema={"row_idx": pl.UInt32, "order_id": pl.String, "dimension": pl.String, "detail": pl.String})
    return pl.DataFrame(rows).sort(["row_idx", "dimension"])


def quarantine(df: pl.DataFrame, failures: pl.DataFrame) -> tuple[pl.DataFrame, pl.DataFrame]:
    """No change from module 7: separates clean_df from quarantined_df
    based on the row_idx values that appear in 'failures'."""
    bad_idx = failures["row_idx"].unique().to_list()
    indexed = df.with_row_index("row_idx")
    quarantined_df = indexed.filter(pl.col("row_idx").is_in(bad_idx)).drop("row_idx")
    clean_df = indexed.filter(~pl.col("row_idx").is_in(bad_idx)).drop("row_idx")
    return clean_df, quarantined_df


def raise_alert(check_name: str, failure_count: int, sample: list[dict]) -> dict:
    return {
        "alert": "data_quality_incident", "pipeline": "kiosko_orders_s04", "check_name": check_name,
        "run_at": PIPELINE_RUN_AT, "severity": "high" if failure_count >= 5 else "medium",
        "failure_count": failure_count, "sample": sample[:3],
    }

quarantine() and raise_alert() are, literally, module 7's same two functions — zero changes. The only piece that gets adapted is build_failure_report(), and only in the parameter it receives (schema instead of implicitly using OrdersSchema), not in its internal logic.

Step 2 — the complete run

# run_s04_incident.py
import duckdb
import pandera
import polars as pl

from kiosko_trust import (
    PIPELINE_RUN_AT, REFERENCE_PRICES, build_failure_report, contract_to_pandera_schema,
    gate_failure_count, load_contract, quarantine, raise_alert, run_full_gate,
)

pl.Config.set_fmt_str_lengths(60)

print(f"pandera version: {pandera.__version__}\n")

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

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

print(f"Contract: {contract.dataset} v{contract.contract_version}, on_violation={contract.on_violation}")
print(f"Rows read from orders_s04: {s04_df.height}\n")

gate_results = run_full_gate(
    s04_df, dim_product_df, REFERENCE_PRICES, schema,
    run_at=PIPELINE_RUN_AT, sla_hours=contract.sla.freshness_hours,
    min_rows=contract.sla.row_count.min, max_rows=contract.sla.row_count.max,
)

print("=== run_full_gate() on orders_2026-08-14.csv ===")
for r in gate_results:
    print(f"  [{r['status']}] {r['check']:<14} {r['detail']}")

failures = gate_failure_count(gate_results)
print(f"\nFailures: {failures} of {len(gate_results)} checks")
assert failures == 6, f"expected 6 failures, got {failures}"
print("assert failures == 6  -> OK")

# --- The contract says on_violation: quarantine -- we honor it ---
if failures > 0 and contract.on_violation == "quarantine":
    print(f"\n=== on_violation='{contract.on_violation}': quarantining ===")
    failure_report = build_failure_report(s04_df, dim_product_df, REFERENCE_PRICES, schema)
    clean_df, quarantined_df = quarantine(s04_df, failure_report)
    print(f"clean_df: {clean_df.height} rows | quarantined_df: {quarantined_df.height} rows")
    assert clean_df.height == 6 and quarantined_df.height == 6

    alert = raise_alert("s04_full_gate", quarantined_df.height, quarantined_df.select(["order_id"]).to_dicts())
    print(f"\nraise_alert(): severity={alert['severity']}, failure_count={alert['failure_count']}")
    print(f"sample: {[s['order_id'] for s in alert['sample']]}")

What to expect

Running python3 run_s04_incident.py for real, with kiosko.duckdb (orders_s04, dim_product) and orders_contract.yaml in the same folder, pandera==0.32.1:

pandera version: 0.32.1

Contract: orders_s04 v1.0.0, on_violation=quarantine
Rows read from orders_s04: 12

=== run_full_gate() on orders_2026-08-14.csv ===
  [FAIL] completeness   1 physical rows: ['ORD-9503']
  [FAIL] uniqueness     2 physical rows: ['ORD-9502']
  [FAIL] validity       1 physical rows: ['ORD-9507']
  [FAIL] consistency    1 orphan rows: ['ORD-9508']
  [FAIL] accuracy       1 rows outside the baseline: ['ORD-9509']
  [FAIL] freshness      47.58h of 24h SLA
  [PASS] volume         12 rows, range [5, 20]

Failures: 6 of 7 checks
assert failures == 6  -> OK

=== on_violation='quarantine': quarantining ===
clean_df: 6 rows | quarantined_df: 6 rows

raise_alert(): severity=high, failure_count=6
sample: ['ORD-9502', 'ORD-9503', 'ORD-9507']

Read this result with the same care you've already trained in every earlier project in this guide. The gate's seven lines confirm, in a single block, what seven complete modules built separately: ORD-9503 for an empty unit_price (completeness, module 2), ORD-9502 repeated twice (uniqueness, module 2), ORD-9507 with quantity=-1 (validity, module 2), ORD-9508 with product_id="P099" (consistency, module 3), ORD-9509 with unit_price=60.00 (accuracy, module 5), and the whole file arriving 47.58 hours past the 24-hour SLA (freshness, module 6). The only line at PASS is volume — twelve rows, within the [5, 20] range the contract declares, the same deliberate contrast module 6 already confirmed: not everything in S04 is broken.

And, because the contract declared on_violation: quarantine since module 4, the system doesn't stop at the report — it separates the six clean rows from the six broken ones, and structures a high-severity alert (because failure_count=6 >= 5, the same threshold module 7 already set). This is this entire guide's first moment where a policy declared in a YAML contract translates, automatically, into a real action on real data — nobody had to read orders_contract.yaml and decide "this means quarantining"; the code read it on its own.

Table: every failure, its row, its origin module

Check	Status	order_id	Module where the tool got built
completeness	FAIL	ORD-9503	Module 2 (OrdersSchema, generated in M4 from the contract)
uniqueness	FAIL	ORD-9502 (two appearances)	Module 2
validity	FAIL	ORD-9507	Module 2
consistency	FAIL	ORD-9508	Module 3 (validate_referential_integrity())
accuracy	FAIL	ORD-9509	Module 5 (check_price_baseline())
freshness	FAIL	(a file property, not a row's)	Module 6 (check_freshness())
volume	PASS	(a file property, not a row's)	Module 6 (check_volume())

Diagram: from the contract to quarantine, in a single run

flowchart TD
    A["orders_2026-08-14.csv\n12 rows"] --> B["run_full_gate()"]
    B --> C{"6 of 7 FAIL"}
    C -->|"contract.on_violation\n== 'quarantine'"| D["build_failure_report()"]
    D --> E["quarantine()"]
    E --> F["clean_df: 6 rows\ncontinue the normal pipeline"]
    E --> G["quarantined_df: 6 rows\nset aside, not lost"]
    C --> H["raise_alert()\nseverity=high"]

Common mistakes

Thinking quarantine() fixes any of the six broken rows. What happens: someone, seeing clean_df: 6 rows | quarantined_df: 6 rows, expects quarantined_df's six rows to somehow have their problems resolved — ORD-9503's empty unit_price filled in, ORD-9509's price corrected. Why it happens: it's the same common mistake module 7 already warned about quarantine() — "quarantine" sounds, in everyday language, like a step toward a cure. How to spot it: inspect quarantined_df after running this lesson — every row has exactly the same values it had in the original file, including ORD-9503's unit_price=None. How to fix it: quarantine() is, quite deliberately, a containment mechanism, never a repair one — it separates, it doesn't fix. Correcting the source data (contacting S04, requesting a new file) is a human step following after this report, not something the code does on its own.

Confusing 6 of 7 FAIL with "the system failed." What happens: someone reads Failures: 6 of 7 checks and concludes run_full_gate() has a bug, because "a system that fails most of its checks can't be working well." Why it happens: in everyday software language, "failure" usually means "something broke in the code." How to spot it: review this lesson's assert failures == 6 — it passes with no error at all, exactly what was expected. How to fix it: always distinguish between "the code failed" (an exception, an assert that doesn't pass) and "the code correctly reported the data has six problems" — the second is, precisely, the whole system built in this guide's reason to exist. run_full_gate() working perfectly is exactly what produces 6 of 7 FAIL on a file that really has six problems.

Exercises

Exercise 1 — Run run_s04_incident.py yourself, from scratch. With kiosko.duckdb (orders_s04, dim_product) and orders_contract.yaml in a new folder, run this lesson's complete script. Confirm you see exactly 6 failures, severity=high, and 6/6 in quarantine.

See solution

If orders_s04 has module 2's exact twelve rows and dim_product has module 3's exact four rows, the output should reproduce this lesson's exactly: the six checks at FAIL with their correct order_id, volume at PASS, clean_df: 6, quarantined_df: 6, and severity=high. If your result differs, first check orders_contract.yaml has no accidental change from module 4, lesson 3's version — any change to sla.row_count or sla.freshness_hours would alter volume's or freshness's result.

Exercise 2 — Change contract.on_violation to "reject" (with no modification to the YAML file, only in memory) and decide what behavior the system would have. After loading contract with load_contract(), add the line contract.on_violation = "reject" before the quarantine section. Without writing the rejection code yet, describe in 2-3 sentences what should happen in that case, compared to quarantine.

See solution

With on_violation="reject", the expected behavior would be the same data-engineering-foundations-guide already showed in its module 7 with S04: the entire file gets rejected, with no row loaded at all, not even the six genuinely clean ones — the equivalent of status="failed", rows_loaded=0. This deliberately contrasts with quarantine, which does let good rows through. This lesson's code doesn't implement the "reject" branch because Kiosko's real contract declares quarantine — module 4's lesson 3 already explained why that's the correct decision for this case —, but the exercise confirms run_full_gate() and on_violation are independent pieces: the gate reports the same 6 failures no matter what policy the contract declares; what changes is only which action that result triggers.

Exercise 3 — Argue why raise_alert() receives quarantined_df.height (6) as failure_count, instead of gate_failure_count(gate_results) (also 6, same number, different origin). In 2-3 sentences, explain why these two numbers coincide in this specific case, and whether they'd always coincide.

See solution

Both numbers coincide in this case — 6 and 6 — but they count different things: gate_failure_count(gate_results) counts checks that failed (out of a maximum of 7), while quarantined_df.height counts physical rows that ended up in quarantine (out of a maximum of 12). The numeric coincidence is specific to this incident: each of the six row-level checks that failed (completeness, uniqueness, validity, consistency, accuracy, plus the duplicate's second appearance) corresponds to exactly one distinct physical row, with no row having two problems at once. If ORD-9509 had, besides the anomalous price, also a nonexistent product_id, there would still be 6 check failures (accuracy and consistency), but only 5 physical rows affected, not 6 — the two numbers would stop coinciding. raise_alert() uses quarantined_df.height on purpose, because it's the number that matters most to whoever receives the alert: how many business rows fell out of the pipeline, not how many technical rules got violated.

Summary and next step

In this lesson you ran run_full_gate() for the first time on real data: S04's complete file, orders_2026-08-14.csv. The result — 6 failures out of 7 checks, volume at PASS — confirms, in a single run, everything seven earlier modules of this guide diagnosed separately. And, because module 4's contract declared on_violation: quarantine, the system didn't stop at the report: it separated the six clean rows from the six broken ones, and structured a high-severity alert, with nobody having to manually translate the contract's policy into an action.

Before moving on you should be able to: name the six checks that fail on S04 and their matching order_id; and explain why volume is the only check at PASS, even though the file has six broken rows.

Lesson 5 runs this exact same system, with no change at all, on a completely different file: a clean day, rebuilt from Kiosko's canonical week. That's where the complete system demonstrates the other half of its promise — that it doesn't just catch what's wrong, it also lets what's right through, with no friction at all.

Resources

• Module 7, lessons 3-4, of this same guide — the exact source of quarantine() and raise_alert(), reused with no change at all in this lesson. src/guides/data-reliability-and-governance-guide/workbook/module-07-the-incident-and-data-governance/en/. In English.
• Module 1, lesson 6, of this same guide — the literal source of orders_2026-08-14.csv's twelve rows this lesson runs again, now through the complete system. src/guides/data-reliability-and-governance-guide/workbook/module-01-when-green-does-not-mean-correct/en/06-running-the-old-quality-gate-on-s04.md. In English.
• This guide's DESIGN. src/guides/data-reliability-and-governance-guide/DISENO.md. In Spanish.