Project: S04's incident response and access policy

Description

This project closes the module. You have build_failure_report() (lesson 2), quarantine() (lesson 3), raise_alert() and the written runbook.md (lesson 4), ACCESS_POLICY with apply_access_policy() (lesson 5), mask_pii() (lesson 6), and generate_catalog() (lesson 7). One step remains: a closing script that brings the six pieces together in one place, run end to end against S04's real incident, with a final report confirming, with executed evidence, that Kiosko finally has what it was missing after six modules of pure diagnosis — a system that acts when something fails, and that protects what not everyone should see.

Connection to the module. This project introduces no new concept — it's the synthesis of lessons 2 through 7, applied end to end to the same incident this guide's six earlier modules diagnosed. It closes this guide's complete thread up to here, and leaves S04 ready for the entire ecosystem's final project: module 8, where the complete system — contract, tests, consistency, anomalies, freshness, lineage, incident, and governance — runs twice, over the broken and over the clean.

An analogy: the complete hospital, in a single shift

This module's lessons built, one by one, a hospital's pieces: the emergency room that sorts cases by severity (quarantine()), the alarm that notifies with no dependence on someone watching (raise_alert()), the protocol written on the wall (runbook.md), the record of who can see which file (ACCESS_POLICY), the way to protect a sensitive record even from someone authorized to consult it (mask_pii()), and the complete building's directory (generate_catalog()). This project is that hospital's complete shift: a patient arrives (orders_2026-08-14.csv), gets diagnosed, gets treated according to protocol, and their information gets handled with the correct care — all in a single pass, with the six pieces working together instead of isolated in separate lessons.

The material you need

You need, in this module's same working folder:

module_7_incident_governance/
├── kiosko.duckdb                              (orders_s04 from module 2,
│                                                dim_product from lesson 2,
│                                                customers from lesson 5)
└── s04_incident_and_governance.py             (this project)

If your kiosko.duckdb doesn't have one of these three tables yet: orders_s04 gets built in module 2, lesson 4; dim_product gets rebuilt inside this same script with modules 3 and 5's same four products, so this project is self-sufficient; customers gets built in this module's lesson 5. This project doesn't re-explain any of those three steps, it assumes you've already done them.

The verified reference solution

# s04_incident_and_governance.py -- module 7 closing project
import hashlib
from datetime import datetime

import duckdb
import pandera.polars as pa
import polars as pl
import yaml

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"


# --- Modules 2, 3, 5, and 6: the reusable tools, no changes ---
class OrdersSchema(pa.DataFrameModel):
    order_id: str = pa.Field(unique=True)
    unit_price: float = pa.Field(nullable=False, ge=0)
    quantity: int = pa.Field(gt=0)


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


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


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


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


# --- Module 7, lessons 2-4: the incident ---
def build_failure_report(df: pl.DataFrame, dim_product_df: pl.DataFrame, reference_prices: dict[str, float]) -> pl.DataFrame:
    rows: list[dict] = []
    try:
        OrdersSchema.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']})"})
    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, lessons 5-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)
    con = duckdb.connect("kiosko.duckdb")

    df = con.sql("SELECT * FROM orders_s04").pl()
    dim_product_df = con.sql("SELECT * FROM dim_product").pl()
    customers_df = con.sql("SELECT * FROM customers").pl()

    print("=== Kiosko: S04's incident and governance (module 7, close) ===\n")

    print("--- 1. The incident: detection, containment, alert ---")
    failures = build_failure_report(df, dim_product_df, REFERENCE_PRICES)
    clean_df, quarantined_df = quarantine(df, failures)
    print(f"failures: {failures.height} physical rows | clean_df: {clean_df.height} | quarantined_df: {quarantined_df.height}")

    sample = failures.select(["order_id", "dimension", "detail"]).to_dicts()
    row_alert = raise_alert("s04_full_gate", quarantined_df.height, sample)
    freshness_result = check_freshness(df, run_at=PIPELINE_RUN_AT, sla_hours=24)
    file_alert = raise_alert("freshness", 1, [freshness_result])
    print(f"Row alert:  check_name={row_alert['check_name']} severity={row_alert['severity']} failure_count={row_alert['failure_count']}")
    print(f"File alert: check_name={file_alert['check_name']} severity={file_alert['severity']} status={freshness_result['status']} hours_since_latest={freshness_result['hours_since_latest']}")
    print("runbook.md: 6 steps (Detect, Triage, Contain, Root cause, Fix, Postmortem) -- written in lesson 4, not reprinted here")

    print("\n--- 2. Governance: role-based access + masking (on customers) ---")
    for role in ["analyst", "finance", "support"]:
        view = build_role_view(customers_df, role)
        print(f"  role={role:<8} columns={view.columns}")
    print(f"\n  Example, role=analyst:")
    print(build_role_view(customers_df, "analyst"))

    print("\n--- 3. The catalog ---")
    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)
    print(f"catalog.yaml written: {len(catalog)} tables, {sum(e['n_columns'] for e in catalog)} columns, {total_pii} PII columns documented")

    print("\n=== Module 7 close ===")
    print(f"Incident: {quarantined_df.height} of {df.height} rows in quarantine, {clean_df.height} clean rows continue the normal pipeline")
    print(f"Structured alerts: 2 (row + file), runbook.md written with all 6 steps")
    print(f"Governance: {len(ACCESS_POLICY)} roles with an access policy, PII deterministically masked, {len(catalog)} tables cataloged ({total_pii} PII columns)")


if __name__ == "__main__":
    main()

What to expect (verified by actually running python3 s04_incident_and_governance.py, with kiosko.duckdb containing orders_s04, dim_product, and customers, pandera==0.32.1, polars==1.43.2, duckdb==1.5.5, pyyaml installed):

=== Kiosko: S04's incident and governance (module 7, close) ===

--- 1. The incident: detection, containment, alert ---
failures: 6 physical rows | clean_df: 6 | quarantined_df: 6
Row alert:  check_name=s04_full_gate severity=high failure_count=6
File alert: check_name=freshness severity=medium status=FAIL hours_since_latest=47.58
runbook.md: 6 steps (Detect, Triage, Contain, Root cause, Fix, Postmortem) -- written in lesson 4, not reprinted here

--- 2. Governance: role-based access + masking (on customers) ---
  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']

  Example, role=analyst:
shape: (6, 2)
┌─────────────┬────────────────┐
│ customer_id ┆ customer_email │
│ ---         ┆ ---            │
│ str         ┆ str            │
╞═════════════╪════════════════╡
│ C001        ┆ eb25122b17a2   │
│ C002        ┆ aa87243c2e43   │
│ C003        ┆ 71b31bf92a73   │
│ C004        ┆ 93f1b0b8ab41   │
│ C005        ┆ 3ae5fff0dc4f   │
│ C006        ┆ b07fd99bf511   │
└─────────────┴────────────────┘

--- 3. The catalog ---
catalog.yaml written: 4 tables, 17 columns, 2 PII columns documented

=== Module 7 close ===
Incident: 6 of 12 rows in quarantine, 6 clean rows continue the normal pipeline
Structured alerts: 2 (row + file), runbook.md written with all 6 steps
Governance: 3 roles with an access policy, PII deterministically masked, 4 tables cataloged (2 PII columns)

Read this complete result with the same care you've already trained in the six earlier modules' projects. Section 1 confirms, in a single place, all of lessons 2 through 4's work: six broken physical rows, separated into quarantine with no loss of the six clean ones, with two structured alerts — one high severity for the row-level incident, one medium severity for the late file — and the already-written runbook referenced, not reinvented on every run. Section 2 confirms lessons 5 and 6's work: three roles, three different access levels to the same customers table, with analyst seeing its email column masked while finance and support see it in plain text, exactly the legitimate-need difference that motivated lesson 5's Going deeper section. Section 3 confirms lesson 7's work: a complete catalog, with all of Kiosko's two PII columns correctly identified. And the final close summarizes, in three lines, the complete system this module built — detection modules 1 through 6 already had, plus containment, communication, and governance, which none of them had.

Diagram: where you came from, where you arrived

flowchart TD
    A["Module 6: DIMENSIONS_STILL_OPEN = []\n-- complete detection,\nno action yet"] --> B["Lesson 2:\nbuild_failure_report()\n-- unites M2+M3+M5"]
    B --> C["Lesson 3:\nquarantine() --\n6 clean, 6 in quarantine"]
    C --> D["Lesson 4:\nraise_alert() x2\n+ runbook.md"]
    D --> E["Lesson 5:\ncustomers + ACCESS_POLICY\n-- who sees which column"]
    E --> F["Lesson 6:\nmask_pii() --\ndeterministic, verified"]
    F --> G["Lesson 7:\ngenerate_catalog() --\ncatalog.yaml"]
    G --> H["This project:\nall 6 pieces together,\na single script"]
    H --> I["Module 8:\nthe complete system,\ncontract -> tests -> ... ->\nincident -> governance,\nrun 2 times"]

Closing the module's promise, point by point

Common mistakes

Thinking this project completely "closes" Kiosko's reliability and governance. What happens: someone, satisfied with this project's complete report — six pieces working together, with no error at all —, concludes Kiosko already has a production-ready data system with no pending work left. Why it happens: after seven modules of building, a clean final report feels like the finish line. How to spot it: review exactly what this project's main() runs — by hand, from the terminal, every time someone executes it. None of this runs automatically when a new file arrives; nothing sends a real alert to any channel; nothing applies ACCESS_POLICY to a real SQL query in production. How to fix it: this project demonstrates the mechanism works, with executed evidence — but turning that mechanism into a real production system (orchestrated, with real integrations, with access control applied at the database level) is, explicitly, the work of the sibling guides this guide's module 8 names in its lesson 7 (airflow-and-declarative-orchestration-guide, aws-core-services-guide / cloud-security-and-guardrails-guide).

Modifying ROLES_WITH_RAW_PII to add "analyst" "to simplify testing." What happens: someone, extending this project for their own use, adds "analyst" to the set of roles with plain-text PII, to avoid dealing with hashes while debugging. Why it happens: masked values are harder to read at a glance than the real data, and during development that feels like unnecessary friction. How to spot it: check whether your reason for modifying ROLES_WITH_RAW_PII has to do with the code writer's convenience, instead of a legitimate business reason for the analyst role. How to fix it: lesson 6 already established, with a concrete argument, why analyst has no legitimate reason to see PII in plain text — development convenience is never, on its own, a valid reason to weaken an access policy. If you need to debug with real data, use a separate environment with test data explicitly marked as such, never relaxing the policy protecting real data (or, as in this case, example data simulating real data).

Exercises

Exercise 1 — Run the whole project yourself, from scratch. In a new folder, with kiosko.duckdb containing orders_s04 (module 2) and customers (this module's lesson 5), run python3 s04_incident_and_governance.py. Confirm you see exactly 6 rows in quarantine, two alerts with high and medium severities, and 4 tables in the catalog.

Exercise 2 — Extend the final report with a count of how many roles can see each customers column in plain text, unmasked. Using ACCESS_POLICY, PII_COLUMNS, and ROLES_WITH_RAW_PII already defined, calculate how many of the three roles have plain-text PII access to customer_email, and how many to customer_phone.

def count_raw_pii_access(policy: dict, pii_columns: set, raw_roles: set) -> dict:
    counts = {}
    for column in pii_columns:
        roles_with_column = {role for role, cols in policy.items() if column in cols}
        roles_with_raw_access = roles_with_column & raw_roles
        counts[column] = len(roles_with_raw_access)
    return counts

result = count_raw_pii_access(ACCESS_POLICY, PII_COLUMNS, ROLES_WITH_RAW_PII)
print(result)

{'customer_email': 2, 'customer_phone': 2}

Exercise 3 — Argue whether quarantine(), raise_alert(), apply_access_policy(), and mask_pii() should be four separate functions (as they are in this guide) or a single handle_incident() function that combines them all. In 2-3 sentences, considering how they're used in this project's main(), argue in favor of keeping them separate.

Summary and next step: closing this module

With this project you close module 7, and with it, the final gap between "Kiosko detects its own problems" (modules 1 through 6's complete achievement) and "Kiosko acts on them, and protects what not everyone should see" (this module's achievement). You built quarantine(), the third answer to a failing check — neither rejecting everything, nor passing in silence —; raise_alert() and a complete runbook.md, which turn a detected incident into a concrete human action; and data governance's three pieces — ACCESS_POLICY, mask_pii(), generate_catalog() — which answer, with executed evidence, the question no earlier module in this guide needed to ask: who can see what.

And, along the way, you followed a discipline running through this guide's eight modules: every new piece got built by reusing, never rewriting, earlier modules' work — build_failure_report() combines, with no duplication, modules 2, 3, and 5's exact tools; quarantine() and raise_alert() operate on that same report; ACCESS_POLICY and mask_pii() protect a completely new table with no line of orders's shared schema touched. That composition habit, more than any individual function, is the biggest lesson this module leaves.

Where you go next. Module 8 — this entire guide's capstone — assembles everything, from module 4's contract to this module's governance, into a single system, run twice: once over orders_2026-08-14.csv, reporting exactly the same six failures you already know; once over a clean day rebuilt from Kiosko's canonical week, reporting zero failures — the final proof the complete system doesn't just detect what's broken, it lets what's correct through with no friction. And its final lesson closes the complete ecosystem, naming, one by one, the sibling guides that solve what this trust system deliberately left pending.

Resources

• Pandera — official documentation (SchemaErrors.failure_cases, build_failure_report()'s technical foundation, reused with no changes since module 2). pandera.readthedocs.io. In English.
• Polars — complete official documentation (with_row_index, join, select, map_elements — the complete set of expressions used across this module). docs.pola.rs. In English.
• PyYAML — official documentation (yaml.safe_dump, used for catalog.yaml). pyyaml.org/wiki/PyYAMLDocumentation. In English.
• Module 6, project (lesson 8), of this same guide — the source of the exact state of S04 this module started from (DIMENSIONS_STILL_OPEN: []). 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.
• This guide's DESIGN — the complete map of all eight modules, including the module 8 that follows. src/guides/data-reliability-and-governance-guide/DISENO.md. In Spanish.