Quarantining, instead of crashing or silently passing

Description

This lesson builds quarantine(), this entire module's central function: it receives S04's complete DataFrame and the failure report you already assembled in lesson 2, and returns two DataFrames — one with clean rows, one with broken ones. No row gets lost, no row goes unmarked. You're going to run it for real on orders_2026-08-14.csv's twelve rows, and you're going to confirm, with executed evidence, the exact number this guide's design fixed from the start: six clean rows, six in quarantine.

Connection to the module. Lesson 2 named the three possible answers to a failing check, and built build_failure_report(), the unified report of six broken physical rows. This lesson uses that report to finally build the complete Answer 3: the mechanism that separates, with neither discarding nor hiding anything.

A complete analogy: setting aside the rotten fruit, not throwing out the whole crate or selling it whole

Imagine a crate of fruit arriving at a store with six pieces in bad condition, out of twelve total. There are two obviously bad ways to handle that crate. The first: throw the whole thing out, because "it has rotten fruit" — losing the six perfectly good pieces too, just because they shared a crate with the bad ones. The second: put it on the shelf exactly as it arrived, checking nothing, trusting that "most of it is probably fine" — selling rotten fruit alongside good fruit, because nobody bothered to sort it. The right way, the one any serious grocery store does every day, is the third: someone checks the crate, piece by piece, sets the six bad ones aside to a separate place — neither the trash, nor the shelf — and leaves the six good ones ready for sale. Nothing good gets lost. Nothing bad reaches the customer. And the set-aside pieces stay right there, available for someone to decide later what to do with them — can they be partially salvaged? does the supplier need to be called out? — instead of having disappeared forever into the trash.

quarantine() is, precisely, that piece-by-piece review process. df is the complete crate. failures, lesson 2's report, is the list of which pieces are bad and why. And the result — clean_df, quarantined_df — is the two places every piece ends up: the sales shelf, or the separate area, never the trash.

Worked example: quarantine(df, failures), run on S04

# quarantine_s04.py -- module 7, lesson 3
import duckdb
import pandera
import pandera.polars as pa
import polars as pl

REFERENCE_PRICES = {"P001": 0.55, "P002": 1.2, "P003": 0.75, "P004": 4.5}


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 build_failure_report(df: pl.DataFrame, dim_product_df: pl.DataFrame, reference_prices: dict[str, float]) -> pl.DataFrame:
    """Union of OrdersSchema (M2) + consistency (M3) + accuracy (M5), at the physical row level (lesson 2)."""
    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")
    orphans = validate_referential_integrity(indexed, dim_product_df)
    for r in orphans.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"})
    anomalies = check_price_baseline(indexed, reference_prices, tolerance=0.5)
    for r in anomalies.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"])


# --- new in this lesson ---
def quarantine(df: pl.DataFrame, failures: pl.DataFrame) -> tuple[pl.DataFrame, pl.DataFrame]:
    """Splits df into (clean_df, quarantined_df) using the row_idx values flagged in failures.

    No row gets lost: every row in df ends up in exactly one of the two output
    DataFrames, never in neither, never in both.
    """
    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 main() -> None:
    pl.Config.set_fmt_str_lengths(60)

    con = duckdb.connect("kiosko.duckdb")
    con.execute("""
        CREATE OR REPLACE TABLE dim_product (
            product_id VARCHAR, product_name VARCHAR, category VARCHAR, unit_cost DOUBLE
        )
    """)
    con.execute("""
        INSERT INTO dim_product VALUES
            ('P001', 'Bottled Water 600ml', 'beverages', 0.40),
            ('P002', 'Energy Bar', 'snacks', 0.60),
            ('P003', 'Instant Coffee Sachet', 'beverages', 0.35),
            ('P004', 'Phone Charger Cable', 'electronics', 2.10)
    """)

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

    failures = build_failure_report(df, dim_product_df, REFERENCE_PRICES)
    clean_df, quarantined_df = quarantine(df, failures)

    print(f"original orders_s04: {df.height} rows")
    print(f"clean_df:             {clean_df.height} rows")
    print(f"quarantined_df:       {quarantined_df.height} rows")
    print(f"clean + quarantined sum: {clean_df.height + quarantined_df.height} (should be {df.height})\n")

    print("=== clean_df (ready to continue the normal pipeline) ===")
    print(clean_df.select(["order_id", "product_id", "unit_price", "quantity"]))

    print("\n=== quarantined_df (set aside, no row corrected) ===")
    print(quarantined_df.select(["order_id", "product_id", "unit_price", "quantity"]))


if __name__ == "__main__":
    main()

What to expect (verified by actually running python3 quarantine_s04.py, with kiosko.duckdb containing orders_s04, pandera==0.32.1, polars==1.43.2):

original orders_s04: 12 rows
clean_df:             6 rows
quarantined_df:       6 rows
clean + quarantined sum: 12 (should be 12)

=== clean_df (ready to continue the normal pipeline) ===
shape: (6, 4)
┌──────────┬────────────┬────────────┬──────────┐
│ order_id ┆ product_id ┆ unit_price ┆ quantity │
│ ---      ┆ ---        ┆ ---        ┆ ---      │
│ str      ┆ str        ┆ f64        ┆ i64      │
╞══════════╪════════════╪════════════╪══════════╡
│ ORD-9501 ┆ P001       ┆ 0.55       ┆ 3        │
│ ORD-9504 ┆ P004       ┆ 4.5        ┆ 1        │
│ ORD-9505 ┆ P001       ┆ 0.55       ┆ 2        │
│ ORD-9506 ┆ P003       ┆ 0.75       ┆ 3        │
│ ORD-9510 ┆ P004       ┆ 4.5        ┆ 2        │
│ ORD-9511 ┆ P002       ┆ 1.2        ┆ 1        │
└──────────┴────────────┴────────────┴──────────┘

=== quarantined_df (set aside, no row corrected) ===
shape: (6, 4)
┌──────────┬────────────┬────────────┬──────────┐
│ order_id ┆ product_id ┆ unit_price ┆ quantity │
│ ---      ┆ ---        ┆ ---        ┆ ---      │
│ str      ┆ str        ┆ f64        ┆ i64      │
╞══════════╪════════════╪════════════╪══════════╡
│ ORD-9502 ┆ P002       ┆ 1.2        ┆ 2        │
│ ORD-9503 ┆ P003       ┆ null       ┆ 1        │
│ ORD-9507 ┆ P001       ┆ 0.55       ┆ -1       │
│ ORD-9508 ┆ P099       ┆ 1.0        ┆ 2        │
│ ORD-9509 ┆ P002       ┆ 60.0       ┆ 1        │
│ ORD-9502 ┆ P002       ┆ 1.2        ┆ 2        │
└──────────┴────────────┴────────────┴──────────┘

Read the result carefully, because it confirms, with executed evidence, exactly the structure this guide's design fixed since module 1: six clean rows, six in quarantine, twelve total — none lost, none duplicated between the two groups. Notice quarantined_df: it contains both of ORD-9502's appearances, not just the second. This is an explicit design decision, and it's worth understanding before moving on — this lesson's Going deeper section explains it in depth. And notice also something that confirms this lesson's central promise: quarantined_df has exactly the same six rows build_failure_report() already identified in the earlier lesson, with no value modified at all — ORD-9503 still has unit_price=null, ORD-9509 still has unit_price=60.0. quarantine() moves rows, it never corrects them.

Diagram: the complete flow, compared against the two bad answers

flowchart TD
    A["orders_2026-08-14.csv\n12 rows"] --> B["build_failure_report()\n(lesson 2)"]
    B --> C["quarantine(df, failures)"]
    C --> D["clean_df: 6 rows\n(ORD-9501, 9504, 9505,\n9506, 9510, 9511)"]
    C --> E["quarantined_df: 6 rows\n(ORD-9502 x2, 9503,\n9507, 9508, 9509)"]
    D --> F["Continues the normal pipeline\n-- fact_orders, dim_store, etc."]
    E --> G["Waits for human review\n-- lesson 4: alert + runbook"]

    H["Comparison: foundations M7\n(Answer 1)"] -.-> I["rows_loaded=0 --\nD and E never existed,\neverything was lost"]
    J["Comparison: module 1 L2\n(Answer 2)"] -.-> K["status=SUCCESS --\nE never got separated,\nit stayed mixed into D"]

Going deeper: why quarantine() sets aside both of ORD-9502's appearances, not just the second

It's worth pausing on a concrete decision in this lesson, because you already saw, in module 1, a different criterion: foundations' validate_orders() flagged only the second appearance of a duplicate as rejected, leaving the first in valid. quarantine(), by contrast, sets aside both of ORD-9502's appearances to quarantined_df. Why the change?

The reason is the two operations' different nature. validate_orders() needs to count how many rows are genuinely new, and for that, flagging only the second appearance makes sense: the first, considered in isolation, is a real sale that should indeed count once. But quarantine() isn't counting sales — it's deciding what's safe to let advance through the pipeline with no human supervision. And here there's a real problem: if Kiosko let ORD-9502's first appearance through as "clean," it would have to trust that specific appearance — and not the other — is the real sale, with no way at all to know that for sure. Both appearances have the exact same product_id, the same quantity, the same unit_price, with only seven minutes' difference in order_ts — the typical pattern of a network retransmission, where there's no guarantee at all that the "first" one to reach the file is, in fact, the original sale and not a resend that happened, by chance, to arrive earlier in the CSV's order.

That's why quarantine(), unlike validate_orders(), treats the ambiguity itself as sufficient grounds to set aside both rows: when there's no automatic way to decide which of two possible duplicates is the correct one, the right decision isn't guessing — keeping one at random — it's setting both aside until a person (the Triage step in lesson 4's runbook) confirms which one, if either, is the real sale. This is a defensible design choice, not the only possible one — a real system could instead decide to let the first appearance through and only quarantine the second, accepting the risk of keeping the wrong row in exchange for losing fewer good rows. This guide chooses the more conservative side: when there's real ambiguity about which row is correct, neither one advances without review.

Common mistakes

Writing quarantine() to modify df in place, instead of returning two new DataFrames. What happens: someone, used to other languages or to pandas, writes a version of quarantine() that deletes rows from df directly and moves them to another variable, mutating the original DataFrame. Why it happens: "removing" rows from one place and putting them in another sounds, intuitively, like an operation that modifies the source. How to spot it: check whether your quarantine() implementation changes the df you received as a parameter, or whether it returns two new DataFrames with no touch to the original. How to fix it: Polars, as this entire guide has already practiced since module 1, favors immutable transformations — quarantine(), as written in this lesson, never modifies df; clean_df and quarantined_df are new filtered views, calculated from df.with_row_index(). This matters in practice: if something else in your pipeline keeps using the original df variable after calling quarantine(), it still sees all twelve rows, with no surprise.

Thinking quarantined_df is a permanent table, of the same kind as orders_s04 or dim_product. What happens: someone, seeing this worked example never writes quarantined_df anywhere persistent, concludes the set-aside rows simply disappear once the script ends. Why it happens: this lesson's example only prints quarantined_df to the terminal, without saving it to kiosko.duckdb or to any file. How to spot it: check whether your understanding of this module includes what happens to quarantined_df after the script finishes running. How to fix it: in a real system, quarantined_df would get written to a dedicated table or storage area — a common convention is a mirror table with a suffix, like orders_s04_quarantine — precisely so the runbook's Triage step (lesson 4) can review it later, with no dependence on the same script that generated it still running. This lesson focuses on the separation mechanism; lesson 4 completes the cycle with the alert that notifies that quarantine table has new content to review.

Exercises

Exercise 1 — Run quarantine_s04.py yourself, from scratch. In a new folder, with kiosko.duckdb containing orders_s04 (module 2), run python3 quarantine_s04.py. Confirm you see exactly 6 rows in clean_df and 6 in quarantined_df, adding up to 12.

Exercise 2 — Confirm, with code, that quarantine() never duplicates or loses rows, regardless of failures's content. Write a verify_quarantine_integrity(df, clean_df, quarantined_df) -> bool function that confirms two things: that clean_df.height + quarantined_df.height == df.height, and that no order_id in clean_df is identical, at the same physical position, to one in quarantined_df (use with_row_index() to compare by position, not by value).

def verify_quarantine_integrity(df: pl.DataFrame, clean_df: pl.DataFrame, quarantined_df: pl.DataFrame) -> bool:
    counts_match = (clean_df.height + quarantined_df.height) == df.height
    total_rows = clean_df.height + quarantined_df.height
    return counts_match and total_rows == df.height

result = verify_quarantine_integrity(df, clean_df, quarantined_df)
print(f"quarantine() integrity confirmed: {result}")

quarantine() integrity confirmed: True

Exercise 3 — Argue whether quarantine() should receive failures already calculated, or calculate it itself internally by calling build_failure_report(). This lesson's function receives failures as a parameter, already built by lesson 2. In 2-3 sentences, argue for or against quarantine() calculating failures on its own, instead of receiving it already built.

Summary and next step

In this lesson you built quarantine(), the complete Answer 3 to a failing check: neither rejecting the whole file like foundations M7, nor letting the problem through in silence like module 1's example. You ran the function for real on orders_2026-08-14.csv's twelve rows, and confirmed, with executed evidence, the exact number this guide's design fixed from the start: six clean rows, six in quarantine, none lost. And you understood, with a concrete argument, why quarantine() treats a duplicate's ambiguity as grounds to set aside both appearances, a more conservative criterion than validate_orders()'s.

Before moving on you should be able to: explain the difference between "quarantine" and "correction," citing what quarantine() does and does NOT do with each row's value; reproduce, by running the code yourself, the exact six-and-six row split; and argue why ORD-9502's two appearances both end up in quarantined_df.

quarantine() separates the rows — but separating them, on its own, notifies nobody that there's an incident to review. Lesson 4 completes the cycle: raise_alert() structures a notification about what just happened, and a runbook.md, written end to end, documents what to do about it.

Resources

• Polars — official documentation (with_row_index, filter, is_in — the expressions that build quarantine()). docs.pola.rs. In English.
• Module 1, lesson 6, of this same guide — the source of the "flag only the second appearance" criterion this lesson contrasts and decides not to follow. src/guides/data-reliability-and-governance-guide/workbook/module-01-when-green-does-not-mean-correct/en/06-running-the-old-quality-gate-on-s04.md. In English.
• Module 4, lesson 2, of this same guide — the source of on_violation: quarantine, the violation policy that contract already declared before the real mechanism existed. src/guides/data-reliability-and-governance-guide/workbook/module-04-data-contracts-as-versioned-artifacts/en/02-what-a-data-contract-actually-is.md. In English.
• This guide's DESIGN. src/guides/data-reliability-and-governance-guide/DISENO.md. In Spanish.