Python Data Validation for Data Engineers: pydantic, pandera, Great Expectations, Schema Enforcement, Row-Level Checks, Data Contracts, and Production Pipeline Patterns

Table of Contents

• Why Validate Data?
• pydantic — Record-Level Schema Enforcement
• pandera — DataFrame-Level Validation
• Great Expectations — Enterprise Data Quality
• pydantic vs pandera vs Great Expectations — When to Use Each
• Production Pattern: Validated ETL Pipeline
• Common Mistakes
• Interview Questions
• Wrapping Up

Our ETL Patterns post introduced validation with a custom DataValidator class. This post goes deeper into the dedicated validation libraries that production pipelines rely on: pydantic for record-level schema enforcement, pandera for DataFrame-level validation, and Great Expectations for comprehensive data quality suites. These tools catch bad data before it reaches your data lake — not after your dashboards start showing wrong numbers.

Analogy — Airport security checkpoints. Your data pipeline is an airport. Raw data is the arriving passengers. Without security (validation), anyone walks through — invalid records, null values, wrong types, duplicate keys. pydantic is the passport check (validates each passenger’s identity document against a schema). pandera is the X-ray machine (scans the entire bag of data for anomalies). Great Expectations is the full security protocol (a documented checklist of every rule the airport enforces, with reports and alerts). You need all three layers for a secure pipeline.

Why Validate Data?

Data quality issues are invisible until they are catastrophic. A null in a join key silently drops rows. A string “N/A” in a numeric column crashes a downstream aggregation. A duplicate primary key inflates your revenue report. Validation catches these issues at the pipeline boundary — before corrupt data propagates through your entire data lake.

Without Validation	With Validation
Bad data silently flows to dashboards	Bad data is caught at ingestion, quarantined, and logged
Schema changes from vendors break pipelines at 3 AM	Schema validation fails fast with a clear error message
“Why is revenue 2x this month?” (duplicate records)	Uniqueness check prevents duplicates from loading
Debugging takes hours (where did the bad data enter?)	Validation reports show exactly which rows failed and why

pydantic — Record-Level Schema Enforcement

pydantic validates individual records (dictionaries, API payloads, config files) against a typed schema. You define a model as a Python class with type annotations, and pydantic validates every field when you create an instance. Invalid data raises a detailed error immediately.

Analogy — A job application form with field-level validation. Each field has rules: name must be a string, age must be a positive integer, email must match a pattern. pydantic is the online form — it checks every field as you submit it and tells you exactly which fields are wrong.

Basic pydantic Models

from pydantic import BaseModel, Field, field_validator
from datetime import datetime
from typing import Optional

class Order(BaseModel):
    order_id: int
    product: str
    amount: float = Field(gt=0, description="Order amount must be positive")
    quantity: int = Field(ge=1, le=10000, description="Between 1 and 10,000")
    region: str
    order_date: datetime
    discount: Optional[float] = 0.0

    @field_validator("region")
    @classmethod
    def region_must_be_valid(cls, v):
        valid_regions = {"Ontario", "Quebec", "Alberta", "BC", "Manitoba"}
        if v not in valid_regions:
            raise ValueError(f"Invalid region: {v}. Must be one of {valid_regions}")
        return v

    @field_validator("product")
    @classmethod
    def product_not_empty(cls, v):
        if not v.strip():
            raise ValueError("Product name cannot be empty")
        return v.strip()

# Valid record -- passes validation
order = Order(
    order_id=1, product="Widget", amount=29.99,
    quantity=10, region="Ontario", order_date="2026-07-19"
)
print(order.model_dump())

# Invalid record -- raises ValidationError with details
from pydantic import ValidationError
try:
    bad_order = Order(
        order_id="abc", product="", amount=-10,
        quantity=0, region="InvalidRegion", order_date="not-a-date"
    )
except ValidationError as e:
    print(e)
    # Shows exactly which fields failed and why

Validating a List of Records

from pydantic import BaseModel, ValidationError
import pandas as pd

class OrderRecord(BaseModel):
    order_id: int
    product: str
    amount: float

def validate_records(records):
    # Validate a list of dicts, separating valid and invalid records.
    valid, invalid = [], []
    for i, record in enumerate(records):
        try:
            validated = OrderRecord(**record)
            valid.append(validated.model_dump())
        except ValidationError as e:
            invalid.append({"row": i, "data": record, "errors": e.errors()})
    return valid, invalid

# From a DataFrame
df = pd.DataFrame({
    "order_id": [1, "bad", 3],
    "product": ["Widget", "Gadget", ""],
    "amount": [29.99, 49.99, -5]
})

valid, invalid = validate_records(df.to_dict("records"))
print(f"Valid: {len(valid)}, Invalid: {len(invalid)}")
for err in invalid:
    print(f"  Row {err['row']}: {err['errors']}")

valid_df = pd.DataFrame(valid)

pydantic for Configuration Validation

from pydantic import BaseModel, Field
from pathlib import Path

class PipelineConfig(BaseModel):
    name: str
    source_path: str
    target_table: str
    batch_size: int = Field(default=10000, ge=100, le=1000000)
    compression: str = Field(default="snappy", pattern="^(snappy|gzip|zstd|none)$")
    dry_run: bool = False

# Load and validate config from YAML
import yaml
with open("pipeline_config.yaml") as f:
    raw = yaml.safe_load(f)

config = PipelineConfig(**raw)  # Validates immediately
print(f"Running pipeline: {config.name}, batch size: {config.batch_size}")

pandera — DataFrame-Level Validation

pandera validates entire DataFrames — column types, value ranges, null constraints, uniqueness, and custom checks. Where pydantic validates one record at a time, pandera validates the whole table at once. It integrates natively with pandas.

Analogy — A quality control inspector at a factory. pydantic checks each widget individually (record by record). pandera checks the entire production batch — are all widgets the right size? Are any duplicates? Is the defect rate within tolerance? Are the required fields present?

Schema Definition and Validation

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

# Define a schema
order_schema = pa.DataFrameSchema({
    "order_id": Column(int, Check.greater_than(0), unique=True, nullable=False),
    "product": Column(str, Check.str_length(min_value=1), nullable=False),
    "amount": Column(float, Check.in_range(0.01, 100000), nullable=False),
    "quantity": Column(int, Check.in_range(1, 10000), nullable=False),
    "region": Column(str, Check.isin(["Ontario", "Quebec", "Alberta", "BC"]), nullable=False),
    "order_date": Column("datetime64[ns]", nullable=False),
})

# Valid DataFrame
df = pd.DataFrame({
    "order_id": [1, 2, 3],
    "product": ["Widget", "Gadget", "Doohickey"],
    "amount": [29.99, 49.99, 9.99],
    "quantity": [10, 20, 5],
    "region": ["Ontario", "Quebec", "Alberta"],
    "order_date": pd.to_datetime(["2026-01-01", "2026-01-02", "2026-01-03"])
})

validated_df = order_schema.validate(df)
print("Validation passed!")

# Invalid DataFrame -- pandera raises SchemaError with details
bad_df = pd.DataFrame({
    "order_id": [1, 1, 3],           # Duplicate!
    "product": ["Widget", "", "X"],   # Empty string!
    "amount": [29.99, -5, 9.99],     # Negative!
    "quantity": [10, 20, 0],          # Zero (below min)!
    "region": ["Ontario", "Invalid", "Alberta"],
    "order_date": pd.to_datetime(["2026-01-01", "2026-01-02", "2026-01-03"])
})

try:
    order_schema.validate(bad_df, lazy=True)  # lazy=True collects ALL errors
except pa.errors.SchemaErrors as e:
    print(f"Validation failed with {len(e.failure_cases)} errors:")
    print(e.failure_cases)
    # Shows every failing row, column, check, and the actual value

pandera with Decorators (Schema-in, Schema-out)

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

input_schema = pa.DataFrameSchema({
    "order_id": Column(int, unique=True),
    "amount": Column(float, Check.greater_than(0)),
    "product": Column(str),
})

output_schema = pa.DataFrameSchema({
    "order_id": Column(int, unique=True),
    "amount": Column(float, Check.greater_than(0)),
    "product": Column(str),
    "amount_tax": Column(float, Check.greater_than(0)),  # New column
})

@pa.check_input(input_schema)
@pa.check_output(output_schema)
def add_tax(df: pd.DataFrame) -> pd.DataFrame:
    # Transform function with validated input and output.
    df = df.copy()
    df["amount_tax"] = df["amount"] * 1.13  # 13% tax
    return df

# If input or output violates the schema, pandera raises an error
result = add_tax(df)

Custom Checks

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

# Custom check: order_date must be within the last 90 days
def recent_date_check(series):
    max_age = pd.Timestamp.now() - pd.Timedelta(days=90)
    return series >= max_age

# Custom check: amount should not exceed 3 standard deviations from mean
def no_outliers(series):
    mean, std = series.mean(), series.std()
    return ((series - mean).abs() <= 3 * std)

schema = pa.DataFrameSchema({
    "order_date": Column("datetime64[ns]", Check(recent_date_check, error="Dates older than 90 days")),
    "amount": Column(float, [
        Check.greater_than(0),
        Check(no_outliers, error="Outlier detected (>3 std from mean)")
    ]),
})

Great Expectations — Enterprise Data Quality

Great Expectations (GX) is a comprehensive data quality framework used in enterprise pipelines. It goes beyond schema validation to provide expectation suites (documented rules), validation results (pass/fail reports), data docs (shareable HTML documentation), and checkpoints (automated validation in pipelines). Think of it as the difference between a smoke test (pandera) and a full QA suite (GX).

Basic Usage

import great_expectations as gx

# Create a context (project configuration)
context = gx.get_context()

# Connect to a data source
datasource = context.data_sources.add_pandas("my_datasource")
data_asset = datasource.add_dataframe_asset("orders")

# Create a batch (the data to validate)
import pandas as pd
df = pd.DataFrame({
    "order_id": [1, 2, 3],
    "amount": [29.99, 49.99, 9.99],
    "region": ["Ontario", "Quebec", "Alberta"]
})
batch = data_asset.add_batch_definition_whole_dataframe("my_batch").get_batch(
    batch_parameters={"dataframe": df}
)

# Create expectations (rules)
suite = context.suites.add(gx.ExpectationSuite(name="order_quality"))
suite.add_expectation(gx.expectations.ExpectColumnValuesToNotBeNull(column="order_id"))
suite.add_expectation(gx.expectations.ExpectColumnValuesToBeUnique(column="order_id"))
suite.add_expectation(gx.expectations.ExpectColumnValuesToBeBetween(
    column="amount", min_value=0.01, max_value=100000
))
suite.add_expectation(gx.expectations.ExpectColumnValuesToBeInSet(
    column="region", value_set=["Ontario", "Quebec", "Alberta", "BC"]
))

# Validate
validation_result = batch.validate(suite)
print(f"Success: {validation_result.success}")
print(f"Results: {len(validation_result.results)} expectations checked")

pydantic vs pandera vs Great Expectations — When to Use Each

Feature	pydantic	pandera	Great Expectations
**Validates**	Individual records (dicts)	DataFrames (tables)	DataFrames + databases + files
**Best for**	API payloads, config files, single records	Pipeline transforms, DataFrame I/O	Enterprise data quality suites
**Schema definition**	Python classes with type hints	DataFrameSchema with Column/Check	Expectation suites (JSON/Python)
**Error reporting**	Per-field validation errors	Per-row, per-column failure cases	HTML data docs, validation results
**Complexity**	Simple (one class per model)	Moderate (schema + checks)	Complex (context, suites, checkpoints)
**Integration**	FastAPI, configs, any Python dict	pandas, PySpark	Airflow, Databricks, any data platform
**Install**	`pip install pydantic`	`pip install pandera`	`pip install great_expectations`

Recommendation: Use pydantic for validating individual records (API responses, config files, event payloads). Use pandera for validating DataFrames in ETL transforms (schema checks before and after each transform). Use Great Expectations for enterprise data quality suites with documentation, alerting, and historical tracking.

Production Pattern: Validated ETL Pipeline

import pandas as pd
import pandera as pa
from pandera import Column, Check
from pydantic import BaseModel, Field
from datetime import datetime

# Step 1: Validate config with pydantic
class PipelineConfig(BaseModel):
    source: str
    target_table: str
    batch_size: int = Field(default=10000, ge=100)

config = PipelineConfig(source="vendor_orders.csv", target_table="orders_clean")

# Step 2: Define input/output schemas with pandera
input_schema = pa.DataFrameSchema({
    "order_id": Column(int, Check.greater_than(0), nullable=False),
    "product": Column(str, nullable=False),
    "amount": Column(float, Check.greater_than(0), nullable=False),
})

output_schema = pa.DataFrameSchema({
    "order_id": Column(int, Check.greater_than(0), unique=True, nullable=False),
    "product": Column(str, Check.str_length(min_value=1), nullable=False),
    "amount": Column(float, Check.greater_than(0), nullable=False),
    "loaded_at": Column("datetime64[ns]", nullable=False),
})

# Step 3: Pipeline with validation at every boundary
def run_validated_pipeline(config):
    # Extract
    raw = pd.read_csv(config.source)

    # Validate input schema
    try:
        input_schema.validate(raw, lazy=True)
        print("Input schema: PASS")
    except pa.errors.SchemaErrors as e:
        print(f"Input schema: FAIL ({len(e.failure_cases)} errors)")
        print(e.failure_cases)
        raise

    # Transform
    df = raw.drop_duplicates(subset=["order_id"], keep="last")
    df["product"] = df["product"].str.strip()
    df["loaded_at"] = pd.Timestamp.now()

    # Validate output schema
    try:
        output_schema.validate(df, lazy=True)
        print("Output schema: PASS")
    except pa.errors.SchemaErrors as e:
        print(f"Output schema: FAIL ({len(e.failure_cases)} errors)")
        raise

    # Load
    print(f"Loading {len(df)} validated rows to {config.target_table}")
    return df

result = run_validated_pipeline(config)

Common Mistakes

1. Validating only at the end of the pipeline. By the time bad data reaches the load step, it may have corrupted intermediate results. Validate after extraction (schema check) and after transformation (data quality check) — two checkpoints, not one.

2. Using pydantic for DataFrame validation. pydantic validates one record at a time in Python — iterating 1 million rows through pydantic is extremely slow. Use pandera for DataFrame validation (it operates on the entire column at once using vectorized pandas operations).

3. Not using lazy=True in pandera. By default, pandera raises on the first error. With lazy=True, it collects all errors across all columns and rows before raising — giving you a complete picture of data quality issues instead of fixing them one at a time.

4. Hardcoding valid values instead of loading them from a reference table. Check.isin(["Ontario", "Quebec"]) breaks when a new province is added. Load valid values from a database or config file: Check.isin(get_valid_regions()).

5. Not testing validation schemas. Your pandera schema is code — it can have bugs. Test it with intentionally valid and invalid DataFrames to verify that it catches what it should and passes what it should.

6. Skipping validation in production for “performance.” Validation adds milliseconds. A data quality incident costs hours or days. The performance overhead of pandera on a 1-million-row DataFrame is typically under 1 second — always validate.

7. Not documenting what each validation checks. Six months from now, someone will ask “why does the pipeline reject orders from region X?” without documentation. pandera supports description on checks. Great Expectations generates HTML data docs automatically. Use them.

8. Treating all validation failures the same. A null in a required column is a hard failure (stop the pipeline). An outlier amount might be a warning (log it, continue). Separate critical checks (raise exception) from warning checks (log and continue).

Interview Questions

Q: What is the difference between pydantic and pandera? A: pydantic validates individual records (dictionaries, API payloads) against a schema defined as a Python class with type hints. It processes one record at a time and is ideal for API input validation and config files. pandera validates entire pandas DataFrames against a schema that defines column types, value ranges, null constraints, and uniqueness rules. It operates on columns using vectorized pandas operations, making it much faster than iterating pydantic over millions of rows. Use pydantic for records, pandera for DataFrames.

Q: What is Great Expectations and when would you use it over pandera? A: Great Expectations is an enterprise data quality framework that provides expectation suites (documented validation rules), validation results (pass/fail reports), data docs (shareable HTML documentation), and checkpoints (automated validation integrated with orchestrators like Airflow). Use it over pandera when you need historical tracking of data quality, shareable documentation for stakeholders, integration with data catalogs, or when operating at enterprise scale across dozens of pipelines.

Q: Where in an ETL pipeline should you validate data? A: At two points minimum. After extraction (validate schema — are the expected columns present with the correct types?). After transformation (validate data quality — are values in range, are keys unique, are required fields populated?). Some teams add a third validation before loading (verify row counts, check for unexpected nulls introduced by joins). Validating only at the end means corrupt data has already propagated through intermediate steps.

Q: How does pandera’s lazy=True work and why is it important? A: By default, pandera stops at the first validation error and raises an exception. With lazy=True, it continues checking all columns and rows, collecting every error into a SchemaErrors exception with a failure_cases DataFrame. This is important because data quality issues often cluster — if one column has nulls, others might too. Seeing all errors at once lets you fix them in one pass instead of discovering them one at a time.

Q: How would you validate API responses before inserting into a database? A: Use pydantic to define a model matching the expected API response schema. Parse each response through the model — pydantic validates types, required fields, value constraints, and custom rules. Valid records go to the load function. Invalid records go to a dead letter table with the validation error details. This catches schema changes from the API provider before they corrupt your database.

Q: What is a data contract and how do validation tools help enforce it? A: A data contract is an agreement between a data producer and consumer about the schema, types, allowed values, freshness, and quality of the data. Validation tools enforce it automatically: pandera schemas define the contract in code (column X must be int, non-null, unique), and the pipeline rejects data that violates the contract. Great Expectations adds documentation and historical tracking — you can prove to stakeholders that the contract has been met for every pipeline run.

Q: How do you handle validation failures in a production pipeline without stopping everything? A: Separate critical failures from warnings. Critical checks (null primary key, wrong schema) should raise exceptions and halt the pipeline — loading bad data is worse than loading no data. Warning checks (outlier values, unexpected categories) should log the issue, quarantine the affected rows to a dead letter table, and continue processing the valid rows. This keeps the pipeline running while preserving visibility into data quality issues.

Wrapping Up

Data validation is the difference between a pipeline that works and a pipeline you can trust. Use pydantic for record-level validation (API payloads, configs), pandera for DataFrame validation (ETL transforms), and Great Expectations for enterprise data quality suites. Validate at every boundary — after extraction and after transformation. Separate hard failures from soft warnings. And document what every check does, because the person debugging a validation failure at 3 AM will thank you.

Related posts: — ETL Patterns (extract, transform, load) — Testing with pytest — Error Handling (try, except, finally) — Working with APIs & HTTP

---

---

Naveen Vuppula is a Senior Data Engineering Consultant and app developer based in Ontario, Canada. He writes about Python, SQL, AWS, Azure, and everything data engineering at DriveDataScience.com.