Validating Document Data with Schemas

Extraction is lossy in ways that do not raise exceptions. A PDF table parses, but one column shifts by a row. A spreadsheet reads, but the date column arrives as text on the third file of the month. An OCR pass returns 1O4.5O instead of 104.50. Every one of those produces a DataFrame with the right shape and wrong contents, and every downstream step — the merge, the aggregate, the report — happily processes it.

A schema is the assertion that turns those silent corruptions into loud failures at the point they enter the pipeline. Ad-hoc checks fail here for a familiar reason: they are written for the errors already seen, live scattered through the transform code, and are quietly dropped when someone refactors. A declared schema is one object that states the contract for the whole dataset, runs as a single call, and reports every violation at once.

Prerequisites

python -m venv .venv
source .venv/bin/activate

pip install "pandas>=2.2,<3" "pandera>=0.20,<0.23" "pydantic>=2.7,<3"

mkdir -p data out/quarantine

pandas==2.2.3
pandera==0.22.1
pydantic==2.9.2

The two libraries answer different questions. pandera validates a table: column dtypes, value ranges, uniqueness, cross-column relationships, and dataframe-level invariants such as "the rows sum to the stated total". pydantic validates a record: one invoice, one form submission, with coercion and nested structure. Document pipelines usually need both — pydantic where records arrive one at a time from a form or an API, pandera once they are a frame.

Step 1: Diagnose What the Data Actually Contains

Write the schema against reality, not against the specification. Profile a few real extractions first.

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

EXTRACT = Path("data/extracted_invoices.csv")

def profile(csv_path: Path) -> pd.DataFrame:
    """Per-column dtype, null share, cardinality and a value sample."""
    if not csv_path.exists():
        raise FileNotFoundError(f"No such file: {csv_path}")
    try:
        frame = pd.read_csv(csv_path, dtype=str)      # read raw: infer nothing
    except Exception as exc:
        raise RuntimeError(f"Could not read {csv_path}: {exc}") from exc

    rows = []
    for column in frame.columns:
        series = frame[column]
        numeric = pd.to_numeric(series, errors="coerce")
        rows.append({
            "column": column,
            "null_pct": round(series.isna().mean() * 100, 1),
            "distinct": series.nunique(dropna=True),
            "numeric_pct": round(numeric.notna().mean() * 100, 1),
            "sample": series.dropna().head(2).tolist(),
        })
    return pd.DataFrame(rows)

if __name__ == "__main__":
    print(profile(EXTRACT).to_string(index=False))

      column  null_pct  distinct  numeric_pct                     sample
  invoice_no       0.0      1204        100.0           [100417, 100418]
    supplier       0.4       311          0.0     [Acme Ltd, Beta Group]
  net_amount       0.0      1198         99.7        [1204.50, 98.00]
     vat_rate       2.1         3        100.0                [20.0, 5.0]
   issued_on       0.0       412          0.0   [2026-07-15, 15/07/2026]

Two problems are already visible. net_amount is 99.7% numeric, so a handful of rows carry something else. And issued_on mixes two date formats, which will produce NaT for one of them under any single-format parse — the situation handled in working with Excel dates and number formats.

Step 2: Declare a pandera Schema

Write the contract as a class. Each field carries a dtype and the checks that must hold, and the class-level configuration decides how strict the shape must be.

# pip install "pandera>=0.20,<0.23" "pandas>=2.2,<3"
import pandas as pd
import pandera.pandas as pa
from pandera.typing import Series

class InvoiceSchema(pa.DataFrameModel):
    """The contract every extracted invoice batch must satisfy."""

    invoice_no: Series[str] = pa.Field(str_matches=r"^\d{6}$", unique=True)
    supplier: Series[str] = pa.Field(nullable=False, str_length={"min_value": 2, "max_value": 120})
    net_amount: Series[float] = pa.Field(ge=0, le=1_000_000)
    vat_rate: Series[float] = pa.Field(isin=[0.0, 5.0, 20.0], nullable=True)
    issued_on: Series[pd.Timestamp] = pa.Field(
        ge=pd.Timestamp("2020-01-01"), le=pd.Timestamp("2030-01-01")
    )

    class Config:
        strict = True        # reject unexpected columns — a shifted extract shows up here
        coerce = True        # convert where safe, fail where not
        ordered = False

    @pa.dataframe_check
    def batch_total_is_positive(cls, frame: pd.DataFrame) -> bool:
        """A batch of invoices whose net total is zero or negative is not credible."""
        return frame["net_amount"].sum() > 0

    @pa.check("supplier")
    def no_placeholder_names(cls, series: Series[str]) -> Series[bool]:
        """Reject extraction artefacts that look like supplier names."""
        return ~series.str.strip().str.lower().isin({"unknown", "n/a", "-", "supplier"})

strict=True is the setting that earns its keep on extracted data. A column that appears because a PDF gained a header, or disappears because a template changed, is exactly the drift that silently misaligns downstream joins — and it fails here, by name, on the first run.

coerce=True converts a numeric string to a float and an ISO date string to a timestamp, but it does not invent values: a row whose amount is "1O4.5O" fails rather than becoming NaN. That distinction is what makes coercion safe to leave on.

Step 3: Validate and Split

Validate with lazy=True so the report lists every violation rather than stopping at the first, then split the frame instead of aborting.

# pip install "pandera>=0.20,<0.23" "pandas>=2.2,<3"
from pathlib import Path
import pandas as pd
import pandera.pandas as pa

def validate_split(frame: pd.DataFrame, schema: type[pa.DataFrameModel]
                   ) -> tuple[pd.DataFrame, pd.DataFrame]:
    """Return (valid rows, invalid rows with a reason column)."""
    try:
        return schema.validate(frame, lazy=True), pd.DataFrame()
    except pa.errors.SchemaErrors as exc:
        failures = exc.failure_cases                     # one row per violation
        bad_index = failures["index"].dropna().unique()

        reasons = (
            failures.dropna(subset=["index"])
            .groupby("index")
            .apply(lambda g: "; ".join(
                f"{row.column}: {row.check}" for row in g.itertuples()
            ), include_groups=False)
        )

        invalid = frame.loc[frame.index.isin(bad_index)].copy()
        invalid["_reason"] = invalid.index.map(reasons)
        valid = frame.loc[~frame.index.isin(bad_index)]

        # Column-level failures (missing column, wrong dtype) have no index and fail the batch
        structural = failures[failures["index"].isna()]
        if not structural.empty:
            raise ValueError(
                "Schema failed structurally: "
                + "; ".join(f"{r.column}: {r.check}" for r in structural.itertuples())
            ) from exc
        return valid, invalid

if __name__ == "__main__":
    raw = pd.read_csv(Path("data/extracted_invoices.csv"))
    good, bad = validate_split(raw, InvoiceSchema)
    print(f"{len(good):,} valid, {len(bad):,} quarantined")

Distinguishing row failures from structural failures is the design decision that makes this usable. One invoice with a bad amount is a data-quality event: quarantine it and carry on. A missing column is a contract breach: nothing downstream can be trusted, so the batch stops. Treating both the same way either drops a whole batch over one bad row, or reports a total from a frame that lost a column.

Step 4: Validate Records with pydantic

Where documents arrive one at a time — a filled form, a webhook, an OCR result per page — validate the record before it ever becomes a row.

# pip install "pydantic>=2.7,<3"
from datetime import date
from decimal import Decimal
from typing import Annotated

from pydantic import BaseModel, Field, ValidationError, field_validator

class Invoice(BaseModel):
    """One extracted invoice, validated and coerced at the boundary."""

    invoice_no: Annotated[str, Field(pattern=r"^\d{6}$")]
    supplier: Annotated[str, Field(min_length=2, max_length=120)]
    net_amount: Decimal = Field(ge=0, le=1_000_000, decimal_places=2)
    vat_rate: Decimal = Field(default=Decimal("20.0"))
    issued_on: date

    @field_validator("supplier")
    @classmethod
    def strip_and_reject_placeholders(cls, value: str) -> str:
        cleaned = " ".join(value.split())
        if cleaned.lower() in {"unknown", "n/a", "-"}:
            raise ValueError("placeholder supplier name")
        return cleaned

    @field_validator("net_amount", mode="before")
    @classmethod
    def clean_ocr_digits(cls, value):
        """Repair the two OCR confusions that are safe to repair."""
        if isinstance(value, str):
            return value.replace("O", "0").replace("l", "1").replace(",", "").strip("£ ")
        return value

if __name__ == "__main__":
    try:
        invoice = Invoice(invoice_no="100417", supplier="Acme Ltd",
                          net_amount="£1,2O4.50", vat_rate="20.0", issued_on="2026-07-15")
        print(invoice.net_amount)      # Decimal('1204.50')
    except ValidationError as exc:
        print(exc.errors()[0]["msg"])

Using Decimal rather than float for money is deliberate. Floating-point rounding on a few thousand invoices produces totals that differ by pennies from the source system, and a reconciliation that is out by pennies costs more time to investigate than it ever saves.

Repairing O to 0 inside a validator is a judgement call worth making explicitly: it is safe for a field constrained to digits and decimal points, and it would be reckless on a supplier name. Keep such repairs field-scoped and documented.

Edge Cases and Variants

Nullable Columns and Missing Data

nullable=True permits missing values; the check still applies to the values that are present. Combining nullability with a coverage requirement is usually what you actually want:

# pip install "pandera>=0.20,<0.23"
import pandera.pandas as pa
import pandas as pd

class Schema(pa.DataFrameModel):
    vat_rate: pa.typing.Series[float] = pa.Field(isin=[0.0, 5.0, 20.0], nullable=True)

    @pa.dataframe_check
    def vat_coverage(cls, frame: pd.DataFrame) -> bool:
        """At most 5% of rows may be missing a VAT rate."""
        return frame["vat_rate"].isna().mean() <= 0.05

Cross-Column Rules

Relationships between columns are dataframe checks, not field checks:

# pip install "pandera>=0.20,<0.23"
@pa.dataframe_check
def gross_matches_net_plus_vat(cls, frame: pd.DataFrame) -> pd.Series:
    """Row-wise: gross must equal net plus VAT, to the penny."""
    expected = (frame["net_amount"] * (1 + frame["vat_rate"] / 100)).round(2)
    return (frame["gross_amount"] - expected).abs() <= 0.01

Returning a boolean Series rather than a single bool makes it a row-level check, so the failing rows are identified individually and can be quarantined rather than failing the batch.

Schema Versioning

Extraction templates change. Keep the schema in version control beside the extractor, and record which version validated a batch:

# pip install "pandas>=2.2,<3"
SCHEMA_VERSION = "invoices-v3"

valid = valid.assign(_schema_version=SCHEMA_VERSION, _validated_at=pd.Timestamp.utcnow())

That one column answers "which rules did this row pass?" six months later, when the rules have moved on and someone is auditing an old report.

Validation of the Validator

A schema that never fails is either perfect or wrong, and it is rarely perfect. Test it against deliberately broken data.

# pip install "pandera>=0.20,<0.23" "pandas>=2.2,<3" pytest
import pandas as pd
import pytest

GOOD = pd.DataFrame({
    "invoice_no": ["100417", "100418"],
    "supplier": ["Acme Ltd", "Beta Group"],
    "net_amount": [1204.50, 98.00],
    "vat_rate": [20.0, 5.0],
    "issued_on": pd.to_datetime(["2026-07-15", "2026-07-16"]),
})

def test_clean_batch_passes():
    valid, invalid = validate_split(GOOD, InvoiceSchema)
    assert len(valid) == 2 and invalid.empty

@pytest.mark.parametrize("column,value,expected_quarantined", [
    ("net_amount", -5.0, 1),          # negative amount
    ("vat_rate", 17.5, 1),            # rate not in the allowed set
    ("invoice_no", "10041", 1),       # wrong length
    ("supplier", "unknown", 1),       # placeholder name
])
def test_bad_row_is_quarantined(column, value, expected_quarantined):
    frame = GOOD.copy()
    frame.loc[0, column] = value
    valid, invalid = validate_split(frame, InvoiceSchema)
    assert len(invalid) == expected_quarantined
    assert len(valid) == len(GOOD) - expected_quarantined

def test_missing_column_stops_the_batch():
    with pytest.raises(ValueError, match="structurally"):
        validate_split(GOOD.drop(columns=["vat_rate"]), InvoiceSchema)

Parameterised negative tests are what stop a schema rotting. When a new field is added, the test list grows with it, and a rule that was accidentally weakened — ge=0 becoming ge=-1 during a debugging session — fails immediately.

Performance and Scale Notes

pandera validation is vectorised and costs roughly 10–20% of the time taken to read the data. On a million-row frame with a dozen checks, expect a second or two. Three habits keep that from growing.

Validate once, at the boundary, rather than after every transform. Re-validating a frame that has only been filtered wastes the whole cost for no new information.

Prefer built-in field checks (ge, isin, str_matches) over custom checks that use apply, which drops to Python-level iteration. A regular expression check on a million rows runs in milliseconds; the same rule written as a lambda over rows takes minutes.

For chunked ingestion, validate per chunk with the same schema object and accumulate the quarantine. That keeps peak memory flat and gives per-chunk failure counts — combine it with the streaming reader in handling large CSV files with chunking:

# pip install "pandas>=2.2,<3" "pandera>=0.20,<0.23"
def validate_stream(csv_path, schema, chunk_rows: int = 200_000):
    """Validate a large file chunk by chunk, yielding valid frames."""
    quarantined = []
    for chunk in pd.read_csv(csv_path, chunksize=chunk_rows):
        valid, invalid = validate_split(chunk, schema)
        if not invalid.empty:
            quarantined.append(invalid)
        yield valid
    if quarantined:
        pd.concat(quarantined).to_csv("out/quarantine/stream.csv", index=False)

Troubleshooting

Complete Working Script

#!/usr/bin/env python3
"""Validate an extracted dataset against a schema and quarantine what fails.

    pip install "pandas>=2.2,<3" "pandera>=0.20,<0.23"
    python validate_batch.py data/extracted_invoices.csv --quarantine out/quarantine
"""
from __future__ import annotations

import argparse
import sys
from pathlib import Path

import pandas as pd
import pandera.pandas as pa

MAX_QUARANTINE_PCT = 5.0

def load(csv_path: Path) -> pd.DataFrame:
    if not csv_path.exists():
        raise FileNotFoundError(f"No such file: {csv_path}")
    frame = pd.read_csv(csv_path, dtype={"invoice_no": str})
    if "issued_on" in frame.columns:
        frame["issued_on"] = pd.to_datetime(frame["issued_on"], errors="coerce", format="mixed")
    return frame

def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description="Schema-validate an extracted batch.")
    parser.add_argument("source", type=Path)
    parser.add_argument("--quarantine", type=Path, default=Path("out/quarantine"))
    parser.add_argument("--max-quarantine-pct", type=float, default=MAX_QUARANTINE_PCT)
    args = parser.parse_args(argv)

    try:
        raw = load(args.source)
        valid, invalid = validate_split(raw, InvoiceSchema)
    except (FileNotFoundError, ValueError) as exc:
        print(f"error: {exc}", file=sys.stderr)
        return 2                                   # structural failure: stop the pipeline

    share = len(invalid) / max(len(raw), 1) * 100
    print(f"{len(valid):,} valid, {len(invalid):,} quarantined ({share:.2f}%)")

    if not invalid.empty:
        args.quarantine.mkdir(parents=True, exist_ok=True)
        target = args.quarantine / f"{args.source.stem}_rejects.csv"
        invalid.to_csv(target, index=False)
        print(f"rejects written to {target}")

    if share > args.max_quarantine_pct:
        print(f"error: quarantine rate {share:.2f}% exceeds {args.max_quarantine_pct}%",
              file=sys.stderr)
        return 1                                   # too much bad data to publish a report

    out = args.source.with_name(f"{args.source.stem}_validated.parquet")
    valid.to_parquet(out, index=False)
    print(f"wrote {out}")
    return 0

if __name__ == "__main__":
    raise SystemExit(main())

The three distinct exit codes are the interface to whatever schedules this: 0 publishes, 1 means the data is too dirty to report on, and 2 means the extraction itself broke. Wiring those to different alerts is what turns validation into an operational control rather than a log line — see scheduling and logging automation jobs.

FAQ

pandera or pydantic — which should I use? pandera for tables, pydantic for records. A pipeline that extracts one document at a time and then aggregates typically uses pydantic at the extraction boundary and pandera once the records are a frame.

Does coerce=True hide problems? No, it converts where conversion is lossless and fails where it is not. "1204.50" becomes 1204.50; "1O4.5O" fails. What hides problems is errors="coerce" on a manual to_numeric, which turns unparseable values into nulls with no record of what they were.

How strict should strict be?strict=True on anything extracted from a document, because column drift is the most common silent corruption. strict="filter" when a source legitimately adds columns you do not consume.

What quarantine rate is acceptable? Set the threshold from the current baseline and tighten it. A batch that normally quarantines 0.1% and suddenly quarantines 4% has a new problem, even though 4% might be within an absolute limit — alert on the change as well as the level.

Can I generate a schema from existing data? Yes: pandera.infer_schema(frame) produces a starting point. Treat the output as a draft — it encodes the quirks of whatever sample it saw, including the ones you are trying to catch.

Living With a Schema

A schema is a contract with a maintenance cost, and the cost shows up in a predictable place: the day a legitimate business change makes valid data fail. A new VAT rate, a supplier whose name genuinely is one character long, an invoice number that switched from six digits to seven. Each looks like a bug in the pipeline and is not.

Budget for that. Keep the schema in the same repository as the extractor, review changes to it like code, and record why each rule exists — a one-line comment naming the incident that motivated a check is worth more than the check itself six months later. A rule nobody can explain is a rule the next person will delete.

Be equally willing to remove rules. A check that has never fired in a year is either protecting against something that cannot happen or is written so loosely that it never will; both are worth knowing. Reviewing the quarantine store quarterly and asking which rules actually caught something keeps the schema honest and stops it accumulating defensive clutter.

Finally, resist encoding business logic in the schema. Validation answers whether a row is structurally usable. Whether an invoice should have been raised at all is a different question, belongs downstream, and needs a different kind of report.

Related

• Fix pandera SchemaError Dtype Mismatch — the most common validation failure
• Quarantine Invalid Rows in a Pipeline — the reject path in detail
• Extracting PDF Data into pandas — where the frames being validated come from
• Scheduling and Logging Automation Jobs — alerting on validation outcomes
• Generating Reports from Pipeline Data — what validated data feeds

Part of Automating Document Data Pipelines.