Proyecto: la respuesta al incidente y la política de acceso de S04

Descripción

Este proyecto cierra el módulo. Tienes build_failure_report() (lección 2), quarantine() (lección 3), raise_alert() y el runbook.md escrito (lección 4), ACCESS_POLICY con apply_access_policy() (lección 5), mask_pii() (lección 6), y generate_catalog() (lección 7). Falta un solo paso: un script de cierre que junte las seis piezas en un solo lugar, corrido de punta a punta sobre el incidente real de S04, con un reporte final que confirme, con evidencia ejecutada, que Kiosko por fin tiene lo que le faltaba después de seis módulos de puro diagnóstico — un sistema que actúa cuando algo falla, y que protege lo que no todo el mundo debería ver.

Conexión con el módulo. Este proyecto no introduce ningún concepto nuevo — es la síntesis de las lecciones 2 a 7, aplicada de punta a punta al mismo incidente que diagnosticaron los seis módulos anteriores de esta guía. Cierra el hilo completo de esta guía hasta aquí, y deja a S04 lista para el proyecto final de todo el ecosistema: el módulo 8, donde el sistema completo —contrato, tests, consistencia, anomalías, freshness, linaje, incidente y gobierno— corre dos veces, sobre lo roto y sobre lo limpio.

Una analogía: el hospital completo, en un solo turno

Las lecciones de este módulo construyeron, una por una, las piezas de un hospital: la sala de urgencias que separa casos por gravedad (quarantine()), la alarma que avisa sin depender de que alguien esté mirando (raise_alert()), el protocolo escrito en la pared (runbook.md), el registro de quién puede ver qué expediente (ACCESS_POLICY), la forma de proteger un historial sensible incluso para quien tiene autorización de consultarlo (mask_pii()), y el directorio del edificio completo (generate_catalog()). Este proyecto es el turno completo de ese hospital: un paciente llega (orders_2026-08-14.csv), se le diagnostica, se le trata según protocolo, y su información se maneja con el cuidado correcto — todo en una sola pasada, con las seis piezas trabajando juntas en vez de aisladas en lecciones separadas.

El material que necesitas

Necesitas, en la misma carpeta de trabajo de este módulo:

modulo_7_incidente_gobierno/
├── kiosko.duckdb                              (orders_s04 del modulo 2,
│                                                dim_product de la leccion 2,
│                                                customers de la leccion 5)
└── s04_incident_and_governance.py             (este proyecto)

Si tu kiosko.duckdb no tiene alguna de estas tres tablas todavía: orders_s04 se construye en el módulo 2, lección 4; dim_product se reconstruye dentro de este mismo script con los mismos cuatro productos de los módulos 3 y 5, para que este proyecto sea autosuficiente; customers se construye en la lección 5 de este módulo. Este proyecto no vuelve a explicar ninguno de esos tres pasos, los da por hechos.

La solución de referencia, verificada

# s04_incident_and_governance.py -- proyecto de cierre del modulo 7
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"


# --- Modulos 2, 3, 5 y 6: las herramientas reusables, sin cambios ---
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",
    }


# --- Modulo 7, lecciones 2-4: el incidente ---
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']} no existe en 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']} se aleja {round(r['deviation'], 1)}x del precio de referencia ({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],
    }


# --- Modulo 7, lecciones 5-7: el gobierno ---
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: incidente y gobierno de S04 (modulo 7, cierre) ===\n")

    print("--- 1. El incidente: deteccion, contencion, alerta ---")
    failures = build_failure_report(df, dim_product_df, REFERENCE_PRICES)
    clean_df, quarantined_df = quarantine(df, failures)
    print(f"failures: {failures.height} filas fisicas | 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"Alerta de fila:    check_name={row_alert['check_name']} severity={row_alert['severity']} failure_count={row_alert['failure_count']}")
    print(f"Alerta de archivo: 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 pasos (Detectar, Triage, Contener, Causa raiz, Arreglar, Postmortem) -- escrito en la leccion 4, no se re-imprime aqui")

    print("\n--- 2. El gobierno: acceso por rol + masking (sobre customers) ---")
    for role in ["analyst", "finance", "support"]:
        view = build_role_view(customers_df, role)
        print(f"  role={role:<8} columnas={view.columns}")
    print(f"\n  Ejemplo, role=analyst:")
    print(build_role_view(customers_df, "analyst"))

    print("\n--- 3. El catalogo ---")
    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 escrito: {len(catalog)} tablas, {sum(e['n_columns'] for e in catalog)} columnas, {total_pii} columnas PII documentadas")

    print("\n=== Cierre del modulo 7 ===")
    print(f"Incidente: {quarantined_df.height} de {df.height} filas en cuarentena, {clean_df.height} filas limpias siguen el pipeline normal")
    print(f"Alertas estructuradas: 2 (fila + archivo), runbook.md escrito con los 6 pasos")
    print(f"Gobierno: {len(ACCESS_POLICY)} roles con politica de acceso, PII enmascarada de forma deterministica, {len(catalog)} tablas catalogadas ({total_pii} columnas PII)")


if __name__ == "__main__":
    main()

Qué esperar (verificado corriendo python3 s04_incident_and_governance.py real, con kiosko.duckdb conteniendo orders_s04, dim_product y customers, pandera==0.32.1, polars==1.43.2, duckdb==1.5.5, pyyaml instalado):

=== Kiosko: incidente y gobierno de S04 (modulo 7, cierre) ===

--- 1. El incidente: deteccion, contencion, alerta ---
failures: 6 filas fisicas | clean_df: 6 | quarantined_df: 6
Alerta de fila:    check_name=s04_full_gate severity=high failure_count=6
Alerta de archivo: check_name=freshness severity=medium status=FAIL hours_since_latest=47.58
runbook.md: 6 pasos (Detectar, Triage, Contener, Causa raiz, Arreglar, Postmortem) -- escrito en la leccion 4, no se re-imprime aqui

--- 2. El gobierno: acceso por rol + masking (sobre customers) ---
  role=analyst  columnas=['customer_id', 'customer_email']
  role=finance  columnas=['customer_id', 'customer_email', 'customer_phone']
  role=support  columnas=['customer_id', 'customer_email', 'customer_phone']

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

--- 3. El catalogo ---
catalog.yaml escrito: 4 tablas, 17 columnas, 2 columnas PII documentadas

=== Cierre del modulo 7 ===
Incidente: 6 de 12 filas en cuarentena, 6 filas limpias siguen el pipeline normal
Alertas estructuradas: 2 (fila + archivo), runbook.md escrito con los 6 pasos
Gobierno: 3 roles con politica de acceso, PII enmascarada de forma deterministica, 4 tablas catalogadas (2 columnas PII)

Lee este resultado completo con el mismo cuidado que ya entrenaste en los proyectos de los seis módulos anteriores. La sección 1 confirma, en un solo lugar, todo el trabajo de las lecciones 2 a 4: seis filas físicas rotas, separadas en cuarentena sin perder las seis limpias, con dos alertas estructuradas —una de severidad alta para el incidente de fila, una de severidad media para el archivo tardío— y el runbook ya escrito referenciado, no reinventado en cada corrida. La sección 2 confirma el trabajo de las lecciones 5 y 6: tres roles, tres niveles de acceso distintos a la misma tabla customers, con analyst viendo su columna de correo enmascarada mientras finance y support la ven en texto plano, exactamente la diferencia de necesidad legítima que motivó la Profundización de la lección 5. La sección 3 confirma el trabajo de la lección 7: un catálogo completo, con las dos únicas columnas PII de todo Kiosko correctamente identificadas. Y el cierre final resume, en tres líneas, el sistema completo que este módulo construyó — detección que ya tenían los módulos 1 a 6, más contención, comunicación y gobierno, que no tenía ninguno.

Diagrama: de dónde venías, a dónde llegaste

flowchart TD
    A["Modulo 6: DIMENSIONS_STILL_OPEN = []\n-- deteccion completa,\nninguna accion todavia"] --> B["Leccion 2:\nbuild_failure_report()\n-- une M2+M3+M5"]
    B --> C["Leccion 3:\nquarantine() --\n6 limpias, 6 en cuarentena"]
    C --> D["Leccion 4:\nraise_alert() x2\n+ runbook.md"]
    D --> E["Leccion 5:\ncustomers + ACCESS_POLICY\n-- quien ve que columna"]
    E --> F["Leccion 6:\nmask_pii() --\ndeterminista, verificado"]
    F --> G["Leccion 7:\ngenerate_catalog() --\ncatalog.yaml"]
    G --> H["Este proyecto:\nlas 6 piezas juntas,\nun solo script"]
    H --> I["Modulo 8:\nel sistema completo,\ncontrato -> tests -> ... ->\nincidente -> gobierno,\ncorrido 2 veces"]

Cerrando la promesa del módulo, punto por punto

Errores comunes

Pensar que este proyecto "cierra" completamente la confiabilidad y el gobierno de Kiosko. Qué pasa: alguien, satisfecho con el reporte completo de este proyecto —seis piezas trabajando juntas, sin ningún error—, concluye que Kiosko ya tiene un sistema de datos listo para producción sin ningún trabajo pendiente. Por qué pasa: después de siete módulos de construcción, un reporte final limpio se siente como el punto de llegada. Cómo detectarlo: revisa qué corre exactamente el main() de este proyecto — a mano, desde la terminal, cada vez que alguien lo ejecuta. Nada de esto corre automáticamente cuando llega un archivo nuevo; nada envía una alerta real a ningún canal; nada aplica ACCESS_POLICY a una consulta SQL real en producción. Cómo corregirlo: este proyecto demuestra que el mecanismo funciona, con evidencia ejecutada — pero convertir ese mecanismo en un sistema de producción real (orquestado, con integraciones reales, con control de acceso aplicado a nivel de base de datos) es, explícitamente, el trabajo de las guías hermanas que el módulo 8 de esta guía nombra en su lección 7 (airflow-and-declarative-orchestration-guide, aws-core-services-guide / cloud-security-and-guardrails-guide).

Modificar ROLES_WITH_RAW_PII para agregar "analyst" "para simplificar las pruebas". Qué pasa: alguien, al extender este proyecto para su propio uso, agrega "analyst" al conjunto de roles con PII en texto plano, para no tener que lidiar con hashes al depurar. Por qué pasa: los valores enmascarados son más difíciles de leer a simple vista que el dato real, y durante el desarrollo eso se siente como una fricción innecesaria. Cómo detectarlo: revisa si tu razón para modificar ROLES_WITH_RAW_PII tiene que ver con la conveniencia de quien escribe el código, en vez de con un motivo de negocio legítimo del rol analyst. Cómo corregirlo: la lección 6 ya estableció, con un argumento concreto, por qué analyst no tiene un motivo legítimo de ver PII en texto plano — la conveniencia durante el desarrollo nunca es, por sí sola, una razón válida para debilitar una política de acceso. Si necesitas depurar con datos reales, usa un entorno separado con datos de prueba explícitamente marcados como tales, nunca relajando la política que protege datos reales (o, como en este caso, datos de ejemplo que simulan datos reales).

Ejercicios

Ejercicio 1 — Corre el proyecto completo tú mismo, desde cero. En una carpeta nueva, con kiosko.duckdb conteniendo orders_s04 (módulo 2) y customers (lección 5 de este módulo), corre python3 s04_incident_and_governance.py. Confirma que ves exactamente 6 filas en cuarentena, dos alertas con severidades high y medium, y 4 tablas en el catálogo.

Ejercicio 2 — Extiende el reporte final con un conteo de cuántos roles pueden ver cada columna de customers en texto plano, sin enmascarar. Usando ACCESS_POLICY, PII_COLUMNS y ROLES_WITH_RAW_PII ya definidos, calcula cuántos de los tres roles tienen acceso PII en texto plano a customer_email, y cuántos a 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}

Ejercicio 3 — Argumenta si quarantine(), raise_alert(), apply_access_policy() y mask_pii() deberían ser cuatro funciones separadas (como están en esta guía) o una sola función handle_incident() que las combine todas. En 2-3 frases, considerando cómo se usan en main() de este proyecto, argumenta a favor de mantenerlas separadas.

Resumen y siguiente paso: el cierre de este módulo

Con este proyecto cierras el módulo 7, y con él, la brecha final entre "Kiosko detecta sus propios problemas" (el logro completo de los módulos 1 a 6) y "Kiosko actúa sobre ellos, y protege lo que no todo el mundo debería ver" (el logro de este módulo). Construiste quarantine(), la tercera respuesta frente a un check que falla —ni rechazar todo, ni pasar en silencio—; raise_alert() y un runbook.md completo, que convierten un incidente detectado en una acción humana concreta; y las tres piezas de gobierno de datos —ACCESS_POLICY, mask_pii(), generate_catalog()— que responden, con evidencia ejecutada, la pregunta que ningún módulo anterior de esta guía necesitó hacerse: quién puede ver qué.

Y, en el camino, seguiste una disciplina que atravesó los ocho módulos de esta guía: cada pieza nueva se construyó reusando, nunca reescribiendo, el trabajo de los módulos anteriores — build_failure_report() combina, sin duplicar, las herramientas exactas de los módulos 2, 3 y 5; quarantine() y raise_alert() operan sobre ese mismo reporte; ACCESS_POLICY y mask_pii() protegen una tabla completamente nueva sin tocar ni una línea del esquema compartido de orders. Ese hábito de composición, más que cualquier función individual, es la lección más grande que deja este módulo.

Hacia dónde sigues. El módulo 8 —el capstone de esta guía completa— ensambla todo, desde el contrato del módulo 4 hasta el gobierno de este módulo, en un solo sistema, corrido dos veces: una sobre orders_2026-08-14.csv, reportando exactamente los mismos seis fallos que ya conoces; otra sobre un día limpio reconstruido de la semana canónica de Kiosko, reportando cero fallos — la prueba final de que el sistema completo no solo detecta lo roto, deja pasar lo correcto sin fricción. Y su última lección cierra el ecosistema completo, nombrando, una por una, las guías hermanas que resuelven lo que este sistema de confianza dejó pendiente a propósito.

Recursos

• Pandera — documentación oficial (SchemaErrors.failure_cases, base técnica de build_failure_report(), reusada sin cambios desde el módulo 2). pandera.readthedocs.io. En inglés.
• Polars — documentación oficial completa (with_row_index, join, select, map_elements — el conjunto completo de expresiones usadas en este módulo). docs.pola.rs. En inglés.
• PyYAML — documentación oficial (yaml.safe_dump, usada para catalog.yaml). pyyaml.org/wiki/PyYAMLDocumentation. En inglés.
• Módulo 6, proyecto (lección 8), de esta misma guía — fuente del estado exacto de S04 en el que arrancó este módulo (DIMENSIONS_STILL_OPEN: []). src/guides/data-reliability-and-governance-guide/workbook/module-06-freshness-volume-and-lineage/es/08-project-s04s-freshness-volume-and-lineage-report.md. En español.
• DISEÑO de esta guía — el mapa completo de los ocho módulos, incluido el módulo 8 que sigue. src/guides/data-reliability-and-governance-guide/DISENO.md. En español.