Corriendo la compuerta completa contra el incidente de S04

Descripción

La lección 3 confirmó que run_full_gate() funciona, sobre tres filas de juguete con un solo problema deliberado. Esta lección la corre sobre el archivo real: orders_2026-08-14.csv, las mismas doce líneas de S04 que el módulo 1 diagnosticó a mano, el módulo 2 empezó a atrapar con Pandera, el módulo 3 completó con integridad referencial, el módulo 4 formalizó como contrato, el módulo 5 cerró con accuracy, y el módulo 6 terminó de diagnosticar con freshness y volumen. Siete módulos de trabajo, condensados en una sola llamada a una sola función — y el resultado, 6 fallos de 7 checks, es la confirmación final de que el sistema integrado ve exactamente lo mismo que vieron, por separado, los siete módulos anteriores.

Conexión con el módulo. Esta lección no descubre ningún problema nuevo — cada uno de los seis fallos que vas a ver ya lo conoces, con evidencia, desde un módulo anterior. Lo que es nuevo es la forma: un solo reporte, generado por una sola función, honrando además la política on_violation: quarantine que declaró el contrato del módulo 4 desde el primer día.

Una analogía: la misma auditoría, ahora con un solo informe firmado

Los módulos 1 a 7 de esta guía son como siete auditores que revisaron, cada uno por su cuenta, un área distinta de la misma empresa —contabilidad, inventario, cumplimiento normativo—, cada uno entregando su propio informe en fechas distintas. Todos coinciden, en sus hallazgos individuales, sobre los mismos seis problemas. Lo que faltaba era el informe consolidado: un solo documento, firmado una sola vez, que reúne los seis hallazgos de los siete auditores en una sola tabla, con un veredicto final claro. Esta lección es ese informe consolidado, aplicado al incidente de S04.

El material que necesitas

Necesitas, en un directorio de trabajo nuevo:

modulo_8_compuerta_completa/
├── kiosko.duckdb              (orders_s04, dim_product -- ya construidas en M2-M3)
├── orders_contract.yaml       (modulo 4, leccion 3)
└── kiosko_trust.py            (leccion 3 de este modulo)

Si tu kiosko.duckdb no tiene orders_s04 o dim_product todavía, repite el paso 2 de la lección 4 del módulo 2 y el paso 1 de la lección 4 del módulo 3 — esta lección no vuelve a explicar esos pasos, los da por hechos.

Ejemplo trabajado: la corrida real, más cuarentena y alerta

Paso 1 — build_failure_report(), adaptada a recibir el schema del contrato

quarantine() necesita saber, fila por fila, cuáles tienen algún problema — el mismo trabajo que ya hizo build_failure_report() en el módulo 7, con un solo cambio: en vez de importar la clase OrdersSchema escrita a mano, recibe el schema ya generado desde el contrato, exactamente el mismo que usa run_full_gate().

# agregar a 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:
    """Igual que build_failure_report() del modulo 7, con un solo cambio: recibe
    el schema GENERADO desde el contrato (M4) en vez de la clase OrdersSchema
    escrita a mano -- la misma union de completeness/uniqueness/validity/
    consistency/accuracy en una sola tabla de fallas por fila fisica."""
    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']} 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']})"})
    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]:
    """Sin ningun cambio respecto al modulo 7: separa clean_df de quarantined_df
    a partir de los row_idx que aparecen en '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() y raise_alert() son, literalmente, las mismas dos funciones del módulo 7 — cero cambios. La única pieza que se adapta es build_failure_report(), y solo en el parámetro que recibe (schema en vez de implícitamente usar OrdersSchema), no en su lógica interna.

Paso 2 — la corrida completa

# 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"Contrato: {contract.dataset} v{contract.contract_version}, on_violation={contract.on_violation}")
print(f"Filas leidas de 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() sobre 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"\nFallos: {failures} de {len(gate_results)} checks")
assert failures == 6, f"se esperaban 6 fallos, se obtuvieron {failures}"
print("assert failures == 6  -> OK")

# --- El contrato dice on_violation: quarantine -- lo honramos ---
if failures > 0 and contract.on_violation == "quarantine":
    print(f"\n=== on_violation='{contract.on_violation}': poniendo en cuarentena ===")
    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} filas | quarantined_df: {quarantined_df.height} filas")
    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']]}")

Qué esperar

Corriendo python3 run_s04_incident.py real, con kiosko.duckdb (orders_s04, dim_product) y orders_contract.yaml en la misma carpeta, pandera==0.32.1:

pandera version: 0.32.1

Contrato: orders_s04 v1.0.0, on_violation=quarantine
Filas leidas de orders_s04: 12

=== run_full_gate() sobre orders_2026-08-14.csv ===
  [FAIL] completeness   1 filas fisicas: ['ORD-9503']
  [FAIL] uniqueness     2 filas fisicas: ['ORD-9502']
  [FAIL] validity       1 filas fisicas: ['ORD-9507']
  [FAIL] consistency    1 filas huerfanas: ['ORD-9508']
  [FAIL] accuracy       1 filas fuera de la linea base: ['ORD-9509']
  [FAIL] freshness      47.58h de 24h de SLA
  [PASS] volume         12 filas, rango [5, 20]

Fallos: 6 de 7 checks
assert failures == 6  -> OK

=== on_violation='quarantine': poniendo en cuarentena ===
clean_df: 6 filas | quarantined_df: 6 filas

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

Lee este resultado con el mismo cuidado que ya entrenaste en cada proyecto anterior de esta guía. Las siete líneas del gate confirman, en un solo bloque, lo que siete módulos completos construyeron por separado: ORD-9503 por unit_price vacío (completeness, módulo 2), ORD-9502 repetido dos veces (uniqueness, módulo 2), ORD-9507 con quantity=-1 (validity, módulo 2), ORD-9508 con product_id="P099" (consistency, módulo 3), ORD-9509 con unit_price=60.00 (accuracy, módulo 5), y el archivo completo llegando 47.58 horas después del SLA de 24 (freshness, módulo 6). La única línea en PASS es volumen — doce filas, dentro del rango [5, 20] que declara el contrato, el mismo contraste deliberado que ya confirmó el módulo 6: no todo en S04 está roto.

Y, porque el contrato declaró on_violation: quarantine desde el módulo 4, el sistema no se detiene en el reporte — separa las seis filas limpias de las seis rotas, y estructura una alerta de severidad high (porque failure_count=6 >= 5, el mismo umbral que ya fijó el módulo 7). Este es el primer momento de toda la guía en el que la política declarada en un contrato YAML se traduce, de forma automática, en una acción real sobre datos reales — nadie tuvo que leer orders_contract.yaml y decidir "esto significa que hay que poner en cuarentena"; el código lo leyó por sí solo.

Tabla: cada fallo, su fila, su módulo de origen

Check	Estado	order_id	Módulo donde se construyó la herramienta
completeness	FAIL	ORD-9503	Módulo 2 (OrdersSchema, generado en M4 desde el contrato)
uniqueness	FAIL	ORD-9502 (dos apariciones)	Módulo 2
validity	FAIL	ORD-9507	Módulo 2
consistency	FAIL	ORD-9508	Módulo 3 (validate_referential_integrity())
accuracy	FAIL	ORD-9509	Módulo 5 (check_price_baseline())
freshness	FAIL	(propiedad del archivo, no de una fila)	Módulo 6 (check_freshness())
volume	PASS	(propiedad del archivo, no de una fila)	Módulo 6 (check_volume())

Diagrama: del contrato a la cuarentena, en una sola corrida

flowchart TD
    A["orders_2026-08-14.csv\n12 filas"] --> B["run_full_gate()"]
    B --> C{"6 de 7 FAIL"}
    C -->|"contract.on_violation\n== 'quarantine'"| D["build_failure_report()"]
    D --> E["quarantine()"]
    E --> F["clean_df: 6 filas\nsiguen el pipeline normal"]
    E --> G["quarantined_df: 6 filas\napartadas, no perdidas"]
    C --> H["raise_alert()\nseverity=high"]

Errores comunes

Pensar que quarantine() corrige alguna de las seis filas rotas. Qué pasa: alguien, viendo clean_df: 6 filas | quarantined_df: 6 filas, espera que las seis filas de quarantined_df tengan, de alguna forma, sus problemas resueltos —el unit_price vacío de ORD-9503 relleno, el precio de ORD-9509 corregido—. Por qué pasa: es el mismo error común que ya advirtió el módulo 7 sobre quarantine() — "cuarentena" suena, en el lenguaje cotidiano, a un paso hacia una cura. Cómo detectarlo: inspecciona quarantined_df después de correr esta lección — cada fila tiene exactamente los mismos valores que tenía en el archivo original, incluido el unit_price=None de ORD-9503. Cómo corregirlo: quarantine() es, con toda intención, un mecanismo de contención, nunca de reparación — separa, no arregla. Corregir los datos de origen (contactar a S04, pedir un archivo nuevo) es un paso humano que sigue después de este reporte, no algo que el código haga por su cuenta.

Confundir 6 de 7 FAIL con "el sistema falló". Qué pasa: alguien lee Fallos: 6 de 7 checks y concluye que run_full_gate() tiene un error, porque "un sistema que falla la mayoría de sus verificaciones no puede estar funcionando bien". Por qué pasa: en el lenguaje cotidiano de software, "fallo" suele significar "algo se rompió en el código". Cómo detectarlo: revisa el assert failures == 6 de esta lección — pasa sin ningún error, exactamente lo que se esperaba. Cómo corregirlo: distingue siempre entre "el código falló" (una excepción, un assert que no pasa) y "el código reportó, correctamente, que los datos tienen seis problemas" — la segunda es, precisamente, la razón de ser de todo el sistema construido en esta guía. run_full_gate() funcionando perfectamente es exactamente lo que produce 6 de 7 FAIL sobre un archivo que de verdad tiene seis problemas.

Ejercicios

Ejercicio 1 — Corre run_s04_incident.py tú mismo, desde cero. Con kiosko.duckdb (orders_s04, dim_product) y orders_contract.yaml en una carpeta nueva, corre el script completo de esta lección. Confirma que ves exactamente 6 fallos, severity=high, y 6/6 en la cuarentena.

Ver solución

Si orders_s04 tiene las doce filas exactas del módulo 2 y dim_product las cuatro filas exactas del módulo 3, la salida debería reproducir exactamente la de esta lección: los seis checks en FAIL con sus order_id correctos, volumen en PASS, clean_df: 6, quarantined_df: 6, y severity=high. Si tu resultado difiere, revisa primero que orders_contract.yaml no tenga ningún cambio accidental respecto a la versión del módulo 4, lección 3 — cualquier cambio en sla.row_count o sla.freshness_hours alteraría el resultado de volume o freshness.

Ejercicio 2 — Cambia contract.on_violation a "reject" (sin modificar el archivo YAML, solo en memoria) y decide qué comportamiento tendría el sistema. Después de cargar contract con load_contract(), agrega la línea contract.on_violation = "reject" antes de la sección de cuarentena. Sin escribir el código de rechazo todavía, describe en 2-3 frases qué debería pasar en ese caso, comparado con quarantine.

Ver solución

Con on_violation="reject", el comportamiento esperado sería el mismo que ya mostró data-engineering-foundations-guide en su módulo 7 con S04: el archivo completo se rechaza, sin cargar ninguna fila, ni siquiera las seis genuinamente limpias — el equivalente a status="failed", rows_loaded=0. Esto contrasta, de forma deliberada, con quarantine, que sí deja avanzar las filas buenas. El código de esta lección no implementa la rama de "reject" porque el contrato real de Kiosko declara quarantine —la lección 3 del módulo 4 ya explicó por qué esa es la decisión correcta para este caso—, pero el ejercicio confirma que run_full_gate() y on_violation son piezas independientes: el gate reporta los mismos 6 fallos sin importar qué política declare el contrato; lo que cambia es únicamente qué acción dispara ese resultado.

Ejercicio 3 — Argumenta por qué raise_alert() recibe quarantined_df.height (6) como failure_count, en vez de gate_failure_count(gate_results) (también 6, mismo número, distinto origen). En 2-3 frases, explica por qué estos dos números coinciden en este caso específico, y si siempre coincidirían.

Ver solución

Los dos números coinciden en este caso —6 y 6— pero cuentan cosas distintas: gate_failure_count(gate_results) cuenta verificaciones que fallaron (de un máximo de 7), mientras que quarantined_df.height cuenta filas físicas que terminaron en cuarentena (de un máximo de 12). La coincidencia numérica es específica de este incidente: cada una de las seis verificaciones de fila que falló (completeness, uniqueness, validity, consistency, accuracy más la segunda aparición del duplicado) corresponde a exactamente una fila física distinta, sin ninguna fila con dos problemas a la vez. Si ORD-9509 tuviera, además del precio anómalo, también un product_id inexistente, seguiría habiendo 6 fallos de verificación (accuracy y consistency), pero solo 5 filas físicas afectadas, no 6 — los dos números dejarían de coincidir. raise_alert() usa quarantined_df.height a propósito, porque es el número que más le importa a quien recibe la alerta: cuántas filas de negocio quedaron fuera del pipeline, no cuántas reglas técnicas se violaron.

Resumen y siguiente paso

En esta lección corriste run_full_gate() por primera vez sobre datos reales: el archivo completo de S04, orders_2026-08-14.csv. El resultado —6 fallos de 7 checks, volumen en PASS— confirma, en una sola corrida, todo lo que siete módulos anteriores de esta guía diagnosticaron por separado. Y, porque el contrato del módulo 4 declaró on_violation: quarantine, el sistema no se quedó en el reporte: separó las seis filas limpias de las seis rotas, y estructuró una alerta de severidad high, sin que nadie tuviera que traducir manualmente la política del contrato en una acción.

Antes de avanzar deberías poder: nombrar los seis checks que fallan sobre S04 y su order_id correspondiente; y explicar por qué volume es la única verificación en PASS, a pesar de que el archivo tiene seis filas rotas.

La lección 5 corre este mismo sistema, sin ningún cambio, sobre un archivo completamente distinto: un día limpio, reconstruido de la semana canónica de Kiosko. Ahí es donde el sistema completo demuestra la otra mitad de su promesa — que no solo atrapa lo que está mal, también deja pasar lo que está bien, sin ninguna fricción.

Recursos

• Módulo 7, lecciones 3-4, de esta misma guía — fuente exacta de quarantine() y raise_alert(), reusadas sin ningún cambio en esta lección. src/guides/data-reliability-and-governance-guide/workbook/module-07-the-incident-and-data-governance/es/. En español.
• Módulo 1, lección 6, de esta misma guía — fuente literal de las doce filas de orders_2026-08-14.csv que esta lección vuelve a correr, ahora a través del sistema completo. src/guides/data-reliability-and-governance-guide/workbook/module-01-when-green-does-not-mean-correct/es/06-running-the-old-quality-gate-on-s04.md. En español.
• DISEÑO de esta guía. src/guides/data-reliability-and-governance-guide/DISENO.md. En español.