Project: Kiosko's trusted data platform

Description

This project closes module 8 — and with it, closes the entire guide. You have run_full_gate() assembled and tested (lesson 3), run on S04's real incident with 6 failures and quarantine triggered by the contract itself (lesson 4), run on a clean day with 0 failures (lesson 5), the governance layer — lineage, catalog, access, masking — published and independent from the gate's result (lesson 6), and the honest map of the six boundaries this guide leaves to its sibling guides (lesson 7). One step remains: rebuild the complete system, from scratch, in a single script, with automatic asserts confirming every number — the same closing pattern data-modeling-for-analytics-guide, dbt-analytics-engineering-guide, and lakehouse-and-iceberg-guide's capstones already used before, now applied to a complete data trust and governance system.

Connection to the module. This project introduces no new concept — it's the final integration of this module's seven earlier lessons, and of this entire guide's seven earlier modules. It literally picks back up the promise that opened this guide in module 1: turning an unreliable green checkmark into a system that answers, with executed evidence, whether Kiosko's data is correct, who can see it, and what to do the day it stops being correct.

An analogy: the complete platform, presented all at once

This module's lessons 3 through 6 built, one piece at a time, Kiosko's trust system: the complete checklist (lesson 3), the proof it catches what's broken (lesson 4), the proof it lets through what's clean (lesson 5), and published governance (lesson 6). This project is the moment to repeat the whole process, end to end, in a single continuous gesture — the same final integration that already closed seven modules of this guide and three sibling guides' capstones before, now applied to the complete system, not a single piece.

The material: everything this guide built, in one place

You need, in a new working directory:

kiosko_trusted_platform/
├── kiosko.duckdb                      (orders_s04, orders_clean_day,
│                                        dim_product, dim_store, customers)
├── orders_contract.yaml               (module 4, lesson 3)
└── kiosko_trusted_platform.py         (this project)

If your kiosko.duckdb doesn't have one of these five tables yet: orders_s04 gets built in module 2, lesson 4; orders_clean_day gets built in this module's lesson 5, from orders_2026-08-15.csv; dim_product in module 3, lesson 4; dim_store in module 1, lesson 5 (with S04 added); customers in module 7, lesson 5. This project doesn't re-explain any of those five steps, it assumes you've already done them — it's, quite deliberately, the final integration, not another building lesson.

The verified reference solution

# kiosko_trusted_platform.py -- module 8 closing project (and this entire guide's)
# Kiosko's complete trust system, run twice, with automatic assertions
import hashlib
from datetime import datetime
from typing import Literal

import duckdb
import pandera
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}
MASK_SALT = "kiosko-mask-salt-2026"


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


CHECK_TO_DIMENSION = {
    "not_nullable": "completeness",
    "field_uniqueness": "uniqueness",
    "greater_than(0)": "validity",
}


# --- 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, lineage ---
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",
    }


LINEAGE_MAP = {
    "fact_orders.order_id": ["orders.order_id"],
    "fact_orders.store_id": ["orders.store_id"],
    "fact_orders.product_id": ["orders.product_id"],
    "fact_orders.quantity": ["orders.quantity"],
    "fact_orders.unit_price": ["orders.unit_price"],
    "fact_orders.revenue": ["orders.quantity", "orders.unit_price"],
    "fact_orders.order_ts": ["orders.order_ts"],
    "dim_product.product_id": ["products.product_id"],
    "dim_product.product_name": ["products.product_name"],
    "dim_product.category": ["products.category"],
    "dim_product.unit_cost": ["products.unit_cost"],
    "dim_store.store_id": ["stores.store_id"],
    "dim_store.store_name": ["stores.store_name"],
    "dim_store.city": ["stores.city"],
    "dim_store.country": ["stores.city"],
}


# --- Module 8: the complete assembly ---
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]:
    results: list[dict] = []

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

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

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

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

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


# --- Module 7: the incident ---
def build_failure_report(df, dim_product_df, reference_prices, schema) -> pl.DataFrame:
    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]:
    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],
    }


# --- Module 7: governance ---
ACCESS_POLICY: dict[str, list[str]] = {
    "analyst": ["order_id", "store_id", "product_id", "quantity", "unit_price", "order_ts", "customer_id", "customer_email"],
    "finance": ["order_id", "store_id", "product_id", "quantity", "unit_price", "order_ts", "customer_id", "customer_email", "customer_phone"],
    "support": ["order_id", "customer_id", "customer_email", "customer_phone"],
}
PII_COLUMNS = {"customer_email", "customer_phone"}
ROLES_WITH_RAW_PII = {"finance", "support"}


def apply_access_policy(df: pl.DataFrame, role: str, policy: dict = ACCESS_POLICY) -> pl.DataFrame:
    allowed = [c for c in policy[role] if c in df.columns]
    return df.select(allowed)


def mask_pii(df: pl.DataFrame, columns: list[str], salt: str = MASK_SALT) -> pl.DataFrame:
    def _hash(value: str) -> str:
        return hashlib.sha256((salt + value).encode()).hexdigest()[:12]
    exprs = [pl.col(c).map_elements(_hash, return_dtype=pl.String).alias(c) for c in columns]
    return df.with_columns(exprs)


def build_role_view(df: pl.DataFrame, role: str) -> pl.DataFrame:
    view = apply_access_policy(df, role)
    if role not in ROLES_WITH_RAW_PII:
        pii_present = [c for c in view.columns if c in PII_COLUMNS]
        if pii_present:
            view = mask_pii(view, pii_present)
    return view


def generate_catalog(tables: list[dict]) -> list[dict]:
    catalog = []
    for t in tables:
        n_pii = sum(1 for c in t["columns"] if c["sensitivity"] == "pii")
        entry = dict(t)
        entry["n_columns"] = len(t["columns"])
        entry["n_pii_columns"] = n_pii
        catalog.append(entry)
    return catalog


TABLES_SOURCE = [
    {"table": "orders_s04", "owner": "data-engineering@kiosko", "contract": "contracts/orders_contract.yaml",
     "columns": [{"name": "order_id", "sensitivity": "none"}, {"name": "store_id", "sensitivity": "none"},
                 {"name": "product_id", "sensitivity": "none"}, {"name": "quantity", "sensitivity": "none"},
                 {"name": "unit_price", "sensitivity": "none"}, {"name": "order_ts", "sensitivity": "none"}]},
    {"table": "dim_product", "owner": "data-engineering@kiosko", "contract": None,
     "columns": [{"name": "product_id", "sensitivity": "none"}, {"name": "product_name", "sensitivity": "none"},
                 {"name": "category", "sensitivity": "none"}, {"name": "unit_cost", "sensitivity": "none"}]},
    {"table": "dim_store", "owner": "data-engineering@kiosko", "contract": None,
     "columns": [{"name": "store_id", "sensitivity": "none"}, {"name": "store_name", "sensitivity": "none"},
                 {"name": "city", "sensitivity": "none"}, {"name": "country", "sensitivity": "none"}]},
    {"table": "customers", "owner": "delivery-app@kiosko", "contract": None,
     "columns": [{"name": "customer_id", "sensitivity": "none"}, {"name": "customer_phone", "sensitivity": "pii"},
                 {"name": "customer_email", "sensitivity": "pii"}]},
]


def main() -> None:
    pl.Config.set_fmt_str_lengths(60)
    print("=== Kiosko: the complete trust system (module 8, guide close) ===")
    print(f"pandera version: {pandera.__version__}\n")

    con = duckdb.connect("kiosko.duckdb")
    dim_product_df = con.sql("SELECT * FROM dim_product").pl()
    customers_df = con.sql("SELECT * FROM customers").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"SLA: freshness_hours={contract.sla.freshness_hours}, row_count=[{contract.sla.row_count.min}, {contract.sla.row_count.max}]\n")

    # --- RUN 1: S04, the incident ---
    print("--- RUN 1: orders_2026-08-14.csv (S04) ---")
    s04_df = con.sql("SELECT * FROM orders_s04").pl()
    print(f"Rows read: {s04_df.height}")

    s04_gate = 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,
    )
    for r in s04_gate:
        print(f"  [{r['status']}] {r['check']:<14} {r['detail']}")

    s04_failures = gate_failure_count(s04_gate)
    assert s04_failures == 6, f"expected 6 failures on S04, got {s04_failures}"

    s04_report = build_failure_report(s04_df, dim_product_df, REFERENCE_PRICES, schema)
    clean_df, quarantined_df = quarantine(s04_df, s04_report)
    assert clean_df.height == 6 and quarantined_df.height == 6

    s04_alert = raise_alert("s04_full_gate", quarantined_df.height, quarantined_df.select(["order_id"]).to_dicts())
    assert s04_alert["severity"] == "high"
    print(f"Failures: {s04_failures} of {len(s04_gate)} | quarantine: {quarantined_df.height}/{s04_df.height} | alert: {s04_alert['severity']}")

    # --- RUN 2: clean day, S01-S03 ---
    print("\n--- RUN 2: orders_2026-08-15.csv (clean day) ---")
    clean_day_df = con.sql("SELECT * FROM orders_clean_day").pl()
    print(f"Rows read: {clean_day_df.height}")

    clean_gate = run_full_gate(
        clean_day_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,
    )
    for r in clean_gate:
        print(f"  [{r['status']}] {r['check']:<14} {r['detail']}")

    clean_failures = gate_failure_count(clean_gate)
    assert clean_failures == 0, f"expected 0 failures on the clean day, got {clean_failures}"

    clean_report = build_failure_report(clean_day_df, dim_product_df, REFERENCE_PRICES, schema)
    clean_clean_df, clean_quarantined_df = quarantine(clean_day_df, clean_report)
    assert clean_quarantined_df.height == 0
    print(f"Failures: {clean_failures} of {len(clean_gate)} | quarantine: {clean_quarantined_df.height}/{clean_day_df.height} (nothing to separate)")

    # --- GOVERNANCE: lineage + access + masking + catalog ---
    print("\n--- GOVERNANCE: lineage + access + masking + catalog ---")
    assert len(LINEAGE_MAP) == 15

    role_columns = {role: build_role_view(customers_df, role).columns for role in ["analyst", "finance", "support"]}
    assert role_columns["analyst"] == ["customer_id", "customer_email"]
    assert role_columns["finance"] == ["customer_id", "customer_email", "customer_phone"]
    for role, cols in role_columns.items():
        print(f"  role={role:<8} columns={cols}")

    analyst_view_1 = build_role_view(customers_df, "analyst")
    analyst_view_2 = build_role_view(customers_df, "analyst")
    assert analyst_view_1["customer_email"].to_list() == analyst_view_2["customer_email"].to_list()
    assert analyst_view_1["customer_email"][0] != customers_df["customer_email"][0]

    catalog = generate_catalog(TABLES_SOURCE)
    with open("catalog.yaml", "w") as f:
        yaml.safe_dump(catalog, f, sort_keys=False, allow_unicode=True)
    total_pii = sum(e["n_pii_columns"] for e in catalog)
    assert len(catalog) == 4 and total_pii == 2
    print(f"  lineage: {len(LINEAGE_MAP)} columns | catalog: {len(catalog)} tables, {total_pii} PII columns -> catalog.yaml")

    # --- CLOSE ---
    print("\n=== Final summary: Kiosko's complete trust system ===")
    print(f"S04 (orders_2026-08-14.csv):        {s04_failures} failures of {len(s04_gate)} checks -- {quarantined_df.height} rows in quarantine, {s04_alert['severity']} alert")
    print(f"Clean day (orders_2026-08-15.csv):  {clean_failures} failures of {len(clean_gate)} checks -- 0 rows in quarantine")
    print(f"Governance:                          {len(LINEAGE_MAP)} lineage columns, {len(ACCESS_POLICY)} roles, {len(catalog)} tables cataloged")
    print("\nAll assertions passed. The system catches what's broken and lets what's clean through.")


if __name__ == "__main__":
    main()

What to expect (verified by actually running python3 kiosko_trusted_platform.py, with kiosko.duckdb (orders_s04, orders_clean_day, dim_product, dim_store, customers) and orders_contract.yaml in the same folder, pandera==0.32.1, polars==1.43.2, duckdb==1.5.5, pydantic==2.13.4, pyyaml installed):

=== Kiosko: the complete trust system (module 8, guide close) ===
pandera version: 0.32.1

Contract: orders_s04 v1.0.0, on_violation=quarantine
SLA: freshness_hours=24, row_count=[5, 20]

--- RUN 1: orders_2026-08-14.csv (S04) ---
Rows read: 12
  [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 | quarantine: 6/12 | alert: high

--- RUN 2: orders_2026-08-15.csv (clean day) ---
Rows read: 8
  [PASS] completeness   no rows
  [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      23.2h of 24h SLA
  [PASS] volume         8 rows, range [5, 20]
Failures: 0 of 7 | quarantine: 0/8 (nothing to separate)

--- GOVERNANCE: lineage + access + masking + catalog ---
  role=analyst  columns=['customer_id', 'customer_email']
  role=finance  columns=['customer_id', 'customer_email', 'customer_phone']
  role=support  columns=['customer_id', 'customer_email', 'customer_phone']
  lineage: 15 columns | catalog: 4 tables, 2 PII columns -> catalog.yaml

=== Final summary: Kiosko's complete trust system ===
S04 (orders_2026-08-14.csv):        6 failures of 7 checks -- 6 rows in quarantine, high alert
Clean day (orders_2026-08-15.csv):  0 failures of 7 checks -- 0 rows in quarantine
Governance:                          15 lineage columns, 3 roles, 4 tables cataloged

All assertions passed. The system catches what's broken and lets what's clean through.

Read this complete result with the same care you've already trained in every closing project of this guide. Ten asserts, none decorative: they confirm S04 produces exactly 6 failures (no more, no fewer), that the six broken rows end up in quarantine with no loss of the six clean ones, that the alert has the correct severity, that the clean day produces exactly 0 failures and needs no quarantine at all, that lineage maps the fifteen expected columns, that each role sees exactly the columns it should, that mask_pii() is still deterministic (the same email produces the same hash across two different calls) and is still, really, a hash and not the original value, and that the catalog documents the four tables with the two correct PII columns. No number in this result gets claimed with no evidence — each one gets tested, with code that really runs, the same "verify, don't trust" discipline that sustained every closing project in this entire guide.

Diagram: the eight modules, in a single run

flowchart TD
    M1["M1: 6 of 6 dimensions\ndiagnosed by hand"] --> M2["M2: OrdersSchema\ncompleteness/uniqueness/validity"]
    M2 --> M3["M3: validate_referential_\nintegrity() -- consistency"]
    M3 --> M4["M4: contract_to_pandera_\nschema() -- the contract generates M2"]
    M4 --> M5["M5: check_price_baseline()\n-- accuracy"]
    M5 --> M6["M6: check_freshness()/\ncheck_volume()/LINEAGE_MAP"]
    M6 --> M7["M7: quarantine()/raise_alert()/\nACCESS_POLICY/mask_pii()/catalog"]
    M7 --> M8["M8 (this project):\nrun_full_gate() -- EVERYTHING together,\n2 runs, 10 assertions"]

Closing the entire guide's promise, point by point

Common mistakes

Thinking this project "finishes" data reliability and governance work forever. What happens: someone, satisfied with this project's clean result — ten asserts passing with no error at all —, concludes Kiosko already has a data system completely ready for production, with no pending work left. Why it happens: after eight modules of building, such a tidy final result feels like a definitive finish line. How to spot it: review this same module's lesson 7 — it names, with concrete evidence from this same system, six complete boundaries still missing: real orchestration, immutable table history, CDC, infrastructure observability, real IAM, performance and cost at scale. How to fix it: this project demonstrates the reliability and governance mechanism works, with executed evidence end to end — turning that mechanism into a real production platform is, explicitly, the work of the sibling guides lesson 7 names, not this guide's.

Modifying REFERENCE_PRICES or PIPELINE_RUN_AT "to see what happens," with no revert of the change. What happens: someone, exploring the system's behavior, changes one of this project's two fixed values, runs the script, and keeps the modified version — breaking the byte-for-byte reproducibility that's sustained every number in this guide since module 1. Why it happens: experimenting with the values is a natural, valid way to understand how the system reacts (this module's lesson 5, exercise 1 invites exactly that). How to spot it: if your result no longer shows 6 failures on S04 or 0 on the clean day, compare your copy's REFERENCE_PRICES and PIPELINE_RUN_AT against this project's exact values. How to fix it: experimenting is welcome — and instructive — but always in a separate copy of the script, leaving kiosko_trusted_platform.py exactly as this lesson defines it, so the result stays reproducible for anyone who runs it after you.

Exercises

Exercise 1 — Run the whole project yourself, from scratch. In a new folder, with kiosko.duckdb (with all five complete tables) and orders_contract.yaml, run python3 kiosko_trusted_platform.py. Confirm you see all ten asserts pass with no error, and the final summary matches, line by line, this lesson's.

Exercise 2 — Add a new assert confirming no row in clean_df (S04's six clean rows) coincides with any row in quarantined_df. Using both DataFrames' order_id, write a check confirming quarantine()'s partition is complete and has no overlap — every row ends up in one group or the other, never both.

clean_ids = set(clean_df["order_id"].to_list())
quarantined_ids = set(quarantined_df["order_id"].to_list())
assert clean_ids.isdisjoint(quarantined_ids), "a row can't be clean AND in quarantine at the same time"
assert len(clean_ids) + quarantined_df.height == s04_df.height, "the partition must cover all 12 rows with none lost"
print(f"Partition verified: {len(clean_ids)} + {quarantined_df.height} = {s04_df.height}, no overlap")

Partition verified: 6 + 6 = 12, no overlap

Exercise 3 — Argue whether this guide should add a third run, on a file with exactly one failure (neither all broken, nor all clean), as part of this closing project. In 2-3 sentences, considering what this module's lesson 5, exercise 1 already demonstrated, decide whether this third run would add evidence the two current runs don't cover.

Summary and next step: closing this entire guide

With this project you close module 8 — and with it, you close data-reliability-and-governance-guide entirely. You assembled Kiosko's complete trust system into a single entry point, run_full_gate(), run twice with executed evidence identical byte for byte to each earlier lesson's: exactly 6 failures on S04's real incident — completeness, uniqueness, validity, consistency, accuracy, freshness, with volume as the only PASS —, and 0 failures on a clean day rebuilt from Kiosko's canonical week. You confirmed, with quarantine() and raise_alert(), that module 4's contract doesn't just describe a policy — on_violation: quarantine — but that the complete system really honors it. And you published, alongside both runs, the complete governance layer: fifteen-column lineage, verified role-based access, deterministically masked PII, and a four-table catalog.

Eight modules ago, this guide opened with a quote: "The pipeline ran successfully — all green checkmarks," while the data inside was silently wrong. The system you close today is that lie's complete answer: not a more trustworthy checkmark, but a mechanism that proves, with executed, reproducible evidence, exactly what's right, what's wrong, which dimension every finding corresponds to, who can see it, and what to do the day something breaks.

And, with the same honesty that sustained every module in this guide, this module's lesson 7 already named what comes next: these checks run by hand, not scheduled (airflow-and-declarative-orchestration-guide); the table the contract protects doesn't yet have format-level immutable history (lakehouse-and-iceberg-guide); S04's incident arrived as a batch, not as real-time CDC (streaming-with-kafka-and-flink-guide); this guide's observability covers data, infrastructure observability is another discipline (monitoring-observability-guide); and column governance doesn't replace real IAM (aws-core-services-guide/cloud-security-and-guardrails-guide, plus advanced-sql-querying-guide/cost-optimization-caching-guide). None of those six boundaries invalidates what you built — each one assumes you already know it, and builds on that foundation exactly the way this guide built on the eight earlier ones in the ecosystem.

Resources

• Module 7, project (lesson 8), of this same guide — the exact source of quarantine(), raise_alert(), ACCESS_POLICY, mask_pii(), generate_catalog(), all reused with no changes in this project. src/guides/data-reliability-and-governance-guide/workbook/module-07-the-incident-and-data-governance/en/08-project-s04s-incident-response-and-access-policy.md. In English.
• Module 6, project (lesson 8), of this same guide — the exact source of check_freshness(), check_volume(), and LINEAGE_MAP. src/guides/data-reliability-and-governance-guide/workbook/module-06-freshness-volume-and-lineage/en/08-project-s04s-freshness-volume-and-lineage-report.md. In English.
• Pandera — complete official documentation (DataFrameSchema, lazy, SchemaErrors.failure_cases), the gate's first three checks' technical foundation. pandera.readthedocs.io. In English.
• data-modeling-for-analytics-guide, dbt-analytics-engineering-guide, lakehouse-and-iceberg-guide — this ecosystem's three earlier capstones that already used this same "rebuild everything from scratch, with automatic assertions, in the closing project" pattern. Sibling guides in this ecosystem.
• src/paths/data-engineering-ecosystem/VALIDACION.md — the market audit that motivated this entire guide from its DESIGN, cited one last time in this close. Internal repo document. In Spanish.
• This guide's DESIGN — the complete map of all eight modules, now all closed. src/guides/data-reliability-and-governance-guide/DISENO.md. In Spanish.