Python Doc & Data Automation

Quarantine Invalid Rows in a Pipeline

Two lines appear in nearly every extraction script, and both are wrong:

frame = frame.dropna(subset=["net_amount"])          # silently discards rows
frame = pd.read_csv(path, on_bad_lines="skip")        # silently discards lines

Neither raises. Neither logs. A month later the reconciliation is out by £40,000 and nobody can say which rows were dropped, when, or why — because the evidence was destroyed at the moment the decision was made.

Quarantining is the alternative: failing rows leave the main path but are written somewhere durable, with the reason and enough provenance to trace them back to the document they came from.

Root Cause of Silent Loss

A pipeline has exactly three defensible responses to a bad row. Fix it, if the repair is deterministic and safe. Reject the whole batch, if the failure means nothing can be trusted. Or quarantine the row and continue.

Dropping is a fourth option that looks like the third and is not, because it discards the row and the record that it existed. The distinction only becomes visible when someone asks how many rows the pipeline received against how many it published — a question a dropping pipeline cannot answer.

Dropping against quarantining A batch of one thousand rows enters the pipeline. On the dropping path, forty rows fail and vanish, so the output shows nine hundred and sixty rows with no record of the rest and the counts do not reconcile. On the quarantine path, the same forty rows are written to a rejects store with a reason and a source reference, so the input count equals the published count plus the quarantined count and every rejected row can be traced. Dropping Quarantining 1,000 rows in dropna / skip 40 rows vanish 960 published No record of the 40 counts cannot be reconciled 1,000 rows in validate and split 960 published 40 quarantined reason + source 960 + 40 = 1,000 every rejected row is traceable to its document

What a Quarantine Record Must Carry

A rejects file that holds only the bad row is barely better than dropping it. Five fields make a reject actionable.

# pip install "pandas>=2.2,<3"
from dataclasses import dataclass
from datetime import datetime, timezone

@dataclass(frozen=True)
class RejectMetadata:
    """The columns every quarantined row carries alongside its original values."""

    reason: str          # which rule failed, in words: "net_amount: greater_than_or_equal_to(0)"
    stage: str           # where it failed: "extract", "validate", "merge"
    source: str          # the document or file it came from
    source_row: int      # position within that source, for a human to look it up
    quarantined_at: str  # UTC timestamp

def stamp(frame, reason_series, stage: str, source: str):
    """Attach reject metadata to a frame of failing rows."""
    return frame.assign(
        _reason=reason_series,
        _stage=stage,
        _source=source,
        _source_row=frame.index + 2,          # +2: header row plus zero-based index
        _quarantined_at=datetime.now(timezone.utc).isoformat(timespec="seconds"),
    )

_source_row is the field people forget and then need most: it is what lets someone open the original spreadsheet or PDF and look at the row that failed. Computing it as index + 2 assumes a single header row — adjust for files whose header sits lower, as diagnosed in fix unnamed columns in pandas read_excel.

The Implementation

# pip install "pandas>=2.2,<3" "pandera>=0.20,<0.23"
from datetime import datetime, timezone
from pathlib import Path

import pandas as pd
import pandera.pandas as pa

QUARANTINE_DIR = Path("out/quarantine")

def split_and_quarantine(frame: pd.DataFrame, schema: type[pa.DataFrameModel],
                         source: str, stage: str = "validate",
                         quarantine_dir: Path = QUARANTINE_DIR) -> tuple[pd.DataFrame, int]:
    """Validate, write failing rows to the quarantine store, return the valid frame."""
    try:
        return schema.validate(frame, lazy=True), 0
    except pa.errors.SchemaErrors as exc:
        failures = exc.failure_cases
        row_failures = failures.dropna(subset=["index"])

        structural = failures[failures["index"].isna()]
        if not structural.empty:
            # A missing column invalidates every row; quarantining them all is not useful
            raise ValueError(
                "structural failure: "
                + "; ".join(f"{r.column}: {r.check}" for r in structural.itertuples())
            ) from exc

        reasons = (
            row_failures.groupby("index")
            .apply(lambda g: "; ".join(f"{r.column}: {r.check}" for r in g.itertuples()),
                   include_groups=False)
        )
        bad_index = reasons.index
        rejected = frame.loc[frame.index.isin(bad_index)].copy()
        rejected["_reason"] = rejected.index.map(reasons)
        rejected["_stage"] = stage
        rejected["_source"] = source
        rejected["_source_row"] = rejected.index + 2
        rejected["_quarantined_at"] = datetime.now(timezone.utc).isoformat(timespec="seconds")

        quarantine_dir.mkdir(parents=True, exist_ok=True)
        stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S")
        target = quarantine_dir / f"{Path(source).stem}_{stage}_{stamp}.csv"
        rejected.to_csv(target, index=False)

        valid = frame.loc[~frame.index.isin(bad_index)]
        print(f"{len(rejected):,} row(s) quarantined to {target}")
        return valid, len(rejected)

if __name__ == "__main__":
    raw = pd.read_csv("data/extracted_invoices.csv")
    good, rejected_count = split_and_quarantine(raw, InvoiceSchema, source="extracted_invoices.csv")
    print(f"{len(good):,} row(s) continue")

Two design choices are load-bearing. A timestamped filename per run means a reject file is never overwritten, so a Monday morning investigation can see Friday's rejects. And raising on structural failures rather than quarantining every row keeps the quarantine store meaningful — a million rows in quarantine because a column was renamed is noise, not evidence.

Alerting on the Reject Rate

The count only helps if someone sees it. Compare against both an absolute limit and the recent baseline, because a jump from 0.1% to 3% matters even when 3% is "acceptable".

# pip install "pandas>=2.2,<3"
import json
from pathlib import Path

HISTORY = Path("out/quarantine/rate_history.json")

def record_and_check(rate_pct: float, absolute_limit: float = 5.0,
                     relative_multiple: float = 3.0, window: int = 10) -> list[str]:
    """Append this run's reject rate and return any alerts it triggers."""
    history = json.loads(HISTORY.read_text()) if HISTORY.exists() else []
    alerts = []

    if rate_pct > absolute_limit:
        alerts.append(f"reject rate {rate_pct:.2f}% exceeds the absolute limit {absolute_limit}%")

    recent = history[-window:]
    if len(recent) >= 3:
        baseline = sum(recent) / len(recent)
        if baseline > 0 and rate_pct > baseline * relative_multiple:
            alerts.append(
                f"reject rate {rate_pct:.2f}% is {rate_pct / baseline:.1f}x the recent "
                f"baseline of {baseline:.2f}%"
            )

    history.append(round(rate_pct, 4))
    HISTORY.parent.mkdir(parents=True, exist_ok=True)
    HISTORY.write_text(json.dumps(history[-100:]))
    return alerts

if __name__ == "__main__":
    for alert in record_and_check(3.2):
        print("ALERT:", alert)

Wire the alerts into whatever already carries failures — the logging setup in scheduling and logging automation jobs is the natural home. A rate that climbs slowly over weeks is the signal this catches and a fixed threshold misses.

Why a relative threshold catches what an absolute one misses A line chart of reject rate across ten runs. The rate sits near zero point two percent for seven runs, then jumps to three point two percent on run eight. The absolute limit of five percent is never breached, so a fixed threshold would not alert. The relative rule, three times the recent baseline, is breached at run eight and raises the alert. Reject rate across ten runs 0% 3% 5% absolute limit 5% baseline ~0.2% 16x baseline run 7 run 1 run 10 A fixed 5% threshold never fires here — the relative rule alerts on run 7, when the template changed

Reprocessing From Quarantine

Quarantine is only worthwhile if rows can come back. Once the source is fixed, replay the rejects through the same validation.

# pip install "pandas>=2.2,<3"
from pathlib import Path
import pandas as pd

META_COLUMNS = ["_reason", "_stage", "_source", "_source_row", "_quarantined_at"]

def reprocess(quarantine_dir: Path, schema, source_label: str = "reprocess") -> pd.DataFrame:
    """Re-validate every quarantined row; recovered rows are returned, the rest stay put."""
    files = sorted(quarantine_dir.glob("*.csv"))
    if not files:
        print("nothing to reprocess")
        return pd.DataFrame()

    recovered = []
    for path in files:
        frame = pd.read_csv(path)
        original = frame.drop(columns=[c for c in META_COLUMNS if c in frame.columns])
        valid, still_bad = split_and_quarantine(original, schema, source=source_label,
                                                stage="reprocess")
        if len(valid):
            recovered.append(valid)
            print(f"{path.name}: recovered {len(valid):,} of {len(frame):,}")
        if still_bad == 0 and len(valid) == len(frame):
            path.unlink()                      # fully recovered: retire the reject file
    return pd.concat(recovered, ignore_index=True) if recovered else pd.DataFrame()

Deleting a reject file only when every row in it recovers keeps the audit trail honest. A partially recovered file leaves the remaining rows quarantined under a new timestamp, so the history of what failed and when is never rewritten.

Verification

The reconciliation is the whole point: rows in must equal rows published plus rows quarantined.

# pip install "pandas>=2.2,<3"
from pathlib import Path
import pandas as pd

def assert_reconciles(input_rows: int, published: pd.DataFrame,
                      quarantine_dir: Path, run_stamp: str) -> None:
    """Every input row must be either published or quarantined — never neither."""
    rejects = 0
    for path in quarantine_dir.glob(f"*{run_stamp}*.csv"):
        rejects += len(pd.read_csv(path))

    accounted = len(published) + rejects
    assert accounted == input_rows, (
        f"{input_rows:,} row(s) in, {len(published):,} published + {rejects:,} quarantined "
        f"= {accounted:,} — {input_rows - accounted:,} row(s) unaccounted for"
    )

    for path in quarantine_dir.glob(f"*{run_stamp}*.csv"):
        frame = pd.read_csv(path)
        for column in ("_reason", "_source", "_source_row"):
            assert column in frame.columns, f"{path.name}: missing {column}"
            assert frame[column].notna().all(), f"{path.name}: {column} has blanks"
    print(f"reconciled: {len(published):,} published + {rejects:,} quarantined = {input_rows:,}")

Run this assertion at the end of every pipeline execution. It is the check that makes silent loss impossible to ship: a dropna reintroduced during a refactor fails it immediately, with a count of exactly how many rows went missing.

Retaining rejects for as long as their sources A timeline over ninety days. Source documents are retained for the full window. Quarantine files are retained for the same window so a reject can always be traced back to the document that produced it. Reprocessing typically happens within the first two weeks. A note warns that quarantine retention shorter than source retention makes older rejects unresolvable. Match quarantine retention to source retention source documents retained 90 days quarantine files retained 90 days — same window reprocessing first 2 weeks A shorter quarantine window leaves rejects that can never be traced to their document

FAQ

Where should the quarantine store live? Wherever it will survive and be noticed — object storage with a lifecycle rule, a database table, or a dated folder that a person actually opens. A temporary directory is not a quarantine.

How long should rejects be kept? Long enough to reprocess after the source is fixed, which is usually one to three months. Match the retention of the source documents so a reject never outlives the file it came from.

Should quarantined rows be included in reported totals? No, but the count should be reported alongside the totals. "£4.1m across 960 invoices, 40 excluded pending review" is honest; "£4.1m" alone is not.

What if the same rows fail every run? That is a source problem, not a pipeline problem. Alert on repeat offenders — rows whose _source_row and _reason recur — and fix the extractor or the upstream system.

Does this apply to bad lines that never parse? Yes, and it is where dropping is most common. Capture them with the callable form of on_bad_lines, described in fix pandas ParserError tokenizing data, then write them to the same store with _stage="parse".

Who Reads the Quarantine

A quarantine store that nobody opens is an expensive way to delete data. Naming an owner is what turns it into a control.

In practice the owner is rarely the person who wrote the pipeline. Whether a rejected invoice is a data-entry mistake, a genuine edge case or a supplier doing something new is a domain question, and the answer usually lives with whoever owns the source process. Give them a view they can actually use — a spreadsheet export grouped by reason is more useful than a directory of timestamped CSVs — and agree how often they will look at it.

Agree the escalation too. A reject rate that doubles is a pipeline event and should alert immediately; a steady handful of rejects each week is a process improvement to be worked through at whatever pace the business cares about. Conflating the two produces either alert fatigue or silence, and both end with the same outcome: the store stops being read.

Part of Validating Document Data with Schemas.