pandera

# schemas/orders.py — Order data validation schema
import pandera as pa
from pandera.typing import Series, DataFrame
import pandas as pd

class OrderSchema(pa.DataFrameModel):
    """Schema for validated order records.

    Every row represents a single order transaction.
    Validation runs automatically when data enters the pipeline.
    """

    order_id: Series[str] = pa.Field(
        unique=True,
        str_matches=r"^ORD-\d{8}$",       # Format: ORD-00000001
        description="Unique order identifier",
    )

    customer_id: Series[str] = pa.Field(
        nullable=False,
        str_length={"min_value": 1, "max_value": 50},
    )

    amount: Series[float] = pa.Field(
        ge=0.01,                            # Minimum $0.01
        le=100_000,                         # Maximum $100,000 (sanity check)
        description="Order total in USD",
    )

    status: Series[str] = pa.Field(
        isin=["pending", "processing", "completed", "cancelled", "refunded"],
    )

    currency: Series[str] = pa.Field(
        isin=["USD", "EUR", "GBP"],
        default="USD",
    )

    created_at: Series[pd.Timestamp] = pa.Field(
        nullable=False,
        description="Order creation timestamp (UTC)",
    )

    shipped_at: Series[pd.Timestamp] = pa.Field(
        nullable=True,                      # Not all orders are shipped yet
    )

    items_count: Series[int] = pa.Field(
        ge=1,                               # At least one item per order
        le=100,                             # Max 100 items
    )

    # DataFrame-level validation (checks across columns)
    @pa.dataframe_check
    def shipped_after_created(cls, df: pd.DataFrame) -> Series[bool]:
        """Shipped date must be after creation date (when present)."""
        mask = df["shipped_at"].notna()
        result = pd.Series(True, index=df.index)
        result[mask] = df.loc[mask, "shipped_at"] > df.loc[mask, "created_at"]
        return result

    @pa.dataframe_check
    def completed_must_be_shipped(cls, df: pd.DataFrame) -> Series[bool]:
        """Completed orders must have a shipped date."""
        completed = df["status"] == "completed"
        return ~completed | df["shipped_at"].notna()

    class Config:
        strict = True                       # Reject extra columns not in schema
        coerce = True                       # Auto-coerce types (str → int, etc.)
        name = "OrderSchema"
        description = "Validated order records for the analytics pipeline"

# pipelines/orders.py — Data pipeline with validation
import pandera as pa
from pandera.typing import DataFrame
from schemas.orders import OrderSchema

@pa.check_types                              # Validates return type at runtime
def load_orders(filepath: str) -> DataFrame[OrderSchema]:
    """Load and validate order data from a CSV file.

    Args:
        filepath: Path to the CSV file containing order records.

    Returns:
        Validated DataFrame conforming to OrderSchema.

    Raises:
        pa.errors.SchemaError: If validation fails with details of violations.
    """
    df = pd.read_csv(filepath, parse_dates=["created_at", "shipped_at"])
    return df                                # Auto-validated by @check_types


def process_orders(orders: DataFrame[OrderSchema]) -> pd.DataFrame:
    """Process validated orders into daily revenue summary."""
    return (
        orders
        .query("status == 'completed'")
        .groupby(orders["created_at"].dt.date)
        .agg(
            revenue=("amount", "sum"),
            order_count=("order_id", "count"),
            avg_items=("items_count", "mean"),
        )
        .reset_index()
    )


# Usage
try:
    orders = load_orders("data/orders_2026_03.csv")
    summary = process_orders(orders)
    print(f"Processed {len(orders)} orders → {len(summary)} daily summaries")
except pa.errors.SchemaError as err:
    print(f"❌ Validation failed:\n{err.failure_cases}")
    # failure_cases is a DataFrame showing exactly which rows/columns failed

# schemas/custom_checks.py — Reusable validation checks
import pandera as pa
import pandera.extensions as extensions
import numpy as np

@extensions.register_check_method(
    statistics=["threshold"],
    supported_types=pa.Column,
)
def no_outliers_iqr(series: pd.Series, *, threshold: float = 1.5) -> pd.Series:
    """Flag values outside the IQR fence as failures.

    Args:
        series: The column to check.
        threshold: IQR multiplier (1.5 = standard, 3.0 = extreme only).
    """
    q1 = series.quantile(0.25)
    q3 = series.quantile(0.75)
    iqr = q3 - q1
    lower = q1 - threshold * iqr
    upper = q3 + threshold * iqr
    return (series >= lower) & (series <= upper)


# Usage in schema
class MetricsSchema(pa.DataFrameModel):
    revenue: Series[float] = pa.Field(no_outliers_iqr={"threshold": 3.0})
    latency_ms: Series[float] = pa.Field(no_outliers_iqr={"threshold": 1.5})

# schemas/polars_schema.py — Validate Polars DataFrames
import pandera.polars as pa
import polars as pl

class UserSchema(pa.DataFrameModel):
    user_id: int = pa.Field(unique=True, gt=0)
    email: str = pa.Field(str_matches=r"^[\w.-]+@[\w.-]+\.\w+$")
    plan: str = pa.Field(isin=["free", "pro", "enterprise"])
    mrr: float = pa.Field(ge=0)

# Validate a Polars DataFrame
df = pl.read_parquet("users.parquet")
validated = UserSchema.validate(df)       # Returns validated Polars DataFrame

# tests/test_data_quality.py — Data quality tests
import pytest
import pandera as pa
from schemas.orders import OrderSchema

def test_orders_schema_on_sample_data():
    """Verify the schema accepts known-good data."""
    good_data = pd.DataFrame({
        "order_id": ["ORD-00000001", "ORD-00000002"],
        "customer_id": ["cust-1", "cust-2"],
        "amount": [29.99, 149.00],
        "status": ["completed", "pending"],
        "currency": ["USD", "EUR"],
        "created_at": pd.to_datetime(["2026-01-01", "2026-01-02"]),
        "shipped_at": pd.to_datetime(["2026-01-03", pd.NaT]),
        "items_count": [2, 5],
    })
    validated = OrderSchema.validate(good_data)
    assert len(validated) == 2


def test_orders_schema_rejects_negative_amount():
    """Schema must reject orders with negative amounts."""
    bad_data = pd.DataFrame({
        "order_id": ["ORD-00000001"],
        "customer_id": ["cust-1"],
        "amount": [-10.00],              # Invalid: negative
        "status": ["completed"],
        "currency": ["USD"],
        "created_at": pd.to_datetime(["2026-01-01"]),
        "shipped_at": pd.to_datetime(["2026-01-02"]),
        "items_count": [1],
    })
    with pytest.raises(pa.errors.SchemaError):
        OrderSchema.validate(bad_data)

pip install pandera

# With Polars support
pip install "pandera[polars]"

# With hypothesis for property-based testing
pip install "pandera[hypotheses]"

I have a RAG chatbot that answers questions from our docs. Set up Pandera to evaluate answer quality.

We're testing GPT-4o vs Claude on our customer support prompts. Set up a comparison with Pandera.