Building a Data Validation Pipeline with Pydantic v2 and Pandera

Data pipelines fail silently. A wrong type, a None where you expected a string, a float that slipped through as "NaN" — your downstream model or API consumes garbage and you spend two days debugging why production results are off. Pydantic v2 and Pandera together give you validation at both the record level and the dataframe level, with minimal boilerplate. Here's how to combine them in a real pipeline.

Why Two Libraries

Pydantic v2 (rewritten in Rust) handles schema validation for individual objects: API payloads, config files, event records. It is fast — orders of magnitude faster than v1 — and its model_validator and field-level validators let you express complex invariants cleanly.

Pandera fills the gap Pydantic leaves: it validates DataFrames. When you're ingesting CSVs, Parquet files, or query results into pandas or Polars, you need column-level type checks, range constraints, and cross-column dependencies. Pandera gives you that with a declarative schema that works as a decorator or a standalone call.

The pattern: use Pydantic at the ingestion boundary (parsing configs, incoming API records, raw JSON), and Pandera at the analytical layer (validating the resulting DataFrame before passing it to a model or report generator).

Setting Up the Ingestion Schema

Install the dependencies:

pip install "pydantic>=2.0" pandera pandas

Say you're processing user activity events from an API or a JSON feed. The ingestion schema catches structural issues before any data touches your pipeline:

from pydantic import BaseModel, field_validator, model_validator
from datetime import datetime
from typing import Literal

class ActivityEvent(BaseModel):
    user_id: str
    event_type: Literal["click", "view", "purchase"]
    amount: float | None = None
    timestamp: datetime

    @field_validator("user_id")
    @classmethod
    def user_id_not_empty(cls, v: str) -> str:
        if not v.strip():
            raise ValueError("user_id cannot be blank")
        return v.strip()

    @model_validator(mode="after")
    def purchase_requires_amount(self) -> "ActivityEvent":
        if self.event_type == "purchase" and self.amount is None:
            raise ValueError("purchase events must include an amount")
        return self

model_validator runs after all field validators, which is where you enforce cross-field invariants. Pydantic v2's mode="after" means you receive the already-validated model instance, not raw input.

Parsing a batch, collecting failures rather than aborting on the first bad record:

from pydantic import ValidationError

def parse_events(raw: list[dict]) -> tuple[list[ActivityEvent], list[dict]]:
    valid, failed = [], []
    for record in raw:
        try:
            valid.append(ActivityEvent.model_validate(record))
        except ValidationError as e:
            failed.append({"record": record, "errors": e.errors()})
    return valid, failed

In production pipelines, partial success is almost always preferable to a full stop. You log the failures, continue with what you have, and alert if the rejection rate crosses a threshold.

DataFrame Validation with Pandera

Once you have valid individual records, convert them to a DataFrame and apply Pandera before any aggregation or model inference:

import pandera as pa
from pandera import Column, DataFrameSchema, Check
import pandas as pd

activity_schema = DataFrameSchema(
    {
        "user_id": Column(str, Check.str_matches(r"^[a-zA-Z0-9_-]+$")),
        "event_type": Column(str, Check.isin(["click", "view", "purchase"])),
        "amount": Column(float, Check.greater_than(0), nullable=True),
        "timestamp": Column(pa.DateTime),
    },
    checks=Check(
        lambda df: df.loc[df["event_type"] == "purchase", "amount"].notna().all(),
        error="All purchase events must have a non-null amount",
    ),
    coerce=True,
)

def validate_dataframe(events: list[ActivityEvent]) -> pd.DataFrame:
    df = pd.DataFrame([e.model_dump() for e in events])
    try:
        return activity_schema.validate(df, lazy=True)
    except pa.errors.SchemaErrors as e:
        print("Schema validation failures:")
        print(e.failure_cases)
        raise

The lazy=True flag collects all failures instead of stopping at the first one — essential for getting a complete picture rather than playing whack-a-mole with schema errors.

Putting It Together

Here's the full pipeline structure, wired up with logging:

import logging
import json
from pathlib import Path

logger = logging.getLogger(__name__)

def run_pipeline(source_path: Path) -> pd.DataFrame:
    raw = json.loads(source_path.read_text())

    valid_records, failed_records = parse_events(raw)

    if failed_records:
        logger.warning(
            "Rejected %d / %d records at ingestion",
            len(failed_records),
            len(raw),
        )
        for item in failed_records:
            logger.debug("Rejected: %s — %s", item["record"], item["errors"])

    if not valid_records:
        raise RuntimeError("No valid records to process")

    df = validate_dataframe(valid_records)
    logger.info("Pipeline passed: %d records validated", len(df))
    return df

One thing that matters here: fail loudly at the DataFrame validation step, not silently. If Pandera raises, you want the pipeline to halt — a schema violation at this layer means your upstream assumptions about data shape are wrong, and your downstream output will be corrupted.

Structured Error Reporting

Both libraries produce machine-parseable error objects. Pydantic's ValidationError.errors() returns a list of dicts with loc, msg, and type. Pandera's SchemaErrors.failure_cases is a DataFrame you can serialize to JSON or push to a monitoring table.

def format_pydantic_errors(errors: list[dict]) -> str:
    return "; ".join(
        f"{'.'.join(str(x) for x in e['loc'])}: {e['msg']}"
        for e in errors
    )

Structured errors also help detect patterns. Repeated violations on the same field — especially in pipelines that process user-submitted data — often indicate malformed or adversarial input. Enforcing strict input boundaries here is one of the controls that shows up consistently in practical security hardening checklists for production systems.

The Takeaway

Pydantic v2 handles individual record validation — fast, explicit, with clean cross-field logic. Pandera covers the analytical layer, enforcing DataFrame-level invariants that object schemas can't express. Together they give you two hard checkpoints:

1. At ingestion: reject malformed records, collect failures, log context.
2. At the DataFrame boundary: enforce column types, ranges, and cross-column constraints before anything downstream touches the data.

The cost of catching a schema violation early is a log line and a rejected record. The cost of missing one is a corrupted model output, a wrong report, or a week of forensics. Set both checkpoints and enforce them strictly.

---

I run AYI NEDJIMI Consultants, a cybersecurity consulting firm. We publish free security hardening checklists — PDF and Excel.