Schema validation and tests: pandera and pytest

A six-column payment schema finds an old city name and a negative amount. Compare validation with tests and fixtures.

A schema names columns, types and required value rules

Schema validation checks a whole DataFrame against a written contract: which columns it has, the type of each, whether values may be missing, and which values and ranges are allowed. The pandera library lets you write that contract once in Python and can report eligible failures from its declared checks. Tests check the code instead of the data: pytest runs your cleaning functions on inputs you chose and fails when a result changes. You need both, because they catch different things. A test checks the code on the cases it actually executes; a schema checks the incoming data against its declared rules whenever validation is invoked. Neither establishes every possible property of arbitrary code and data. This lesson writes a pandera schema for an invented table of fee payments, validates a file eagerly and lazily, tests a cleaning step with pytest, and shows where each kind of rule belongs.

A schema boundaryschema boundary: The point where a declared table contract is actually validated, such as incoming data or an output handed to a consumer. is where the declared contract is invoked. In pandera 0.34 you import the pandas API as import pandera.pandas as pa and describe each column with pa.Column: its dtype, a list of checks, and options such as unique=True. Columns are not nullable unless you say so, and strict=True makes a column that the schema does not name an error as well. The schema then validates any DataFrame:

"""The same file validated twice: eagerly, which stops at the first problem, and lazily, which collects them all."""
import pandas as pd
import pandera.pandas as pa

from fee_schema import fee_schema

fees = pd.read_csv("fees.csv", parse_dates=["paid_on"])
try:
    fee_schema.validate(fees)
except pa.errors.SchemaError as err:
    print("eager:", err)

try:
    fee_schema.validate(fees, lazy=True)
except pa.errors.SchemaErrors as err:
    cases = err.failure_cases
    print(f"\nlazy: {len(cases)} failure cases")
    print(cases[["column", "check", "failure_case", "index"]].to_string(index=False))

eager: Column 'city' failed element-wise validator number 0: isin(['Chennai', 'Jaipur', 'Kochi', 'Mumbai', 'Pune']) failure cases: Bombay

lazy: 2 failure cases
column                                                  check failure_case  index
  city isin(['Chennai', 'Jaipur', 'Kochi', 'Mumbai', 'Pune'])       Bombay      2
amount                                    in_range(1, 200000)        -4500      4

"""The contract of the fee-payments table, written once and used by every step that reads or writes the table."""
import pandera.pandas as pa

CITIES = ["Chennai", "Jaipur", "Kochi", "Mumbai", "Pune"]
MODES = ["UPI", "card", "netbanking", "cash"]

fee_schema = pa.DataFrameSchema(
    {
        "payment_id": pa.Column(str, pa.Check.str_matches(r"^P\d{5}$"), unique=True),
        "student_id": pa.Column(str),
        "city": pa.Column(str, pa.Check.isin(CITIES)),
        "amount": pa.Column(int, pa.Check.in_range(1, 200_000)),
        "paid_on": pa.Column("datetime64[us]"),
        "mode": pa.Column(str, pa.Check.isin(MODES)),
    },
    strict=True,  # a column the schema does not name is an error too
)

payment_id,student_id,city,amount,paid_on,mode
P10001,S2041,Pune,18500,2026-07-03,UPI
P10002,S2042,Chennai,18500,2026-07-03,card
P10003,S2043,Bombay,9250,2026-07-04,UPI
P10004,S2044,Jaipur,18500,2026-07-05,netbanking
P10005,S2045,Pune,-4500,2026-07-05,UPI
P10006,S2046,Kochi,18500,2026-07-06,cash
P10007,S2047,Mumbai,9250,2026-07-06,card
P10008,S2048,Chennai,18500,2026-07-07,UPI

Eager validation, the default, stops at the first problem and raises a SchemaError: here the old city name Bombay in the third row. Lazy validationlazy validation: Validation that aggregates eligible failures into a report, subject to parsing, coercion and type prerequisites for later checks., validate(df, lazy=True), aggregates eligible failures into SchemaErrors. Coercion or incompatible types can prevent intended later checks; its failure_cases table reports the failures actually collected: the city and also the negative amount of 4,500, a refund typed into the payments column. Use eager validation while exploring, when the first error is all you need. Use lazy validation to collect eligible problems in one run, and inspect parse or coercion failures before relying on later value-check coverage. The resulting failure table can be counted, saved and sent to whoever owns the source.

Validate where data crosses a boundary: right after a file is read, so that a broken export is reported against its source, and right before a table is written or handed on, so that no later step receives a DataFrame that breaks the contract. Keep the schema in one module that every step imports, as fee_schema.py is here, so the rules live in one place.

Retain the original six-column schema while comparing eager and lazy outcomes on its actual file.

Tests use chosen inputs and validation inspects each run

A pytest test is a function whose name starts with test_, in a file whose name starts with test_; pytest finds them, runs them and reports each one. Inside, a plain assert states what must hold, and pandas.testing.assert_frame_equal compares two DataFrames value by value, including dtypes, column names and the index, which is what you want for the output of a cleaning step:

"""Runs the tests with pytest and prints one line per test (pytest's own summary also prints the time taken)."""
import sys

import pytest


class OneLinePerTest:
    def pytest_runtest_logreport(self, report):
        if report.when == "call" or report.failed:
            print(f"{report.nodeid}: {report.outcome}")
            if report.failed:
                print("   ", report.longrepr.reprcrash.message.splitlines()[0])


code = pytest.main(["-p", "no:terminal", "-p", "no:cacheprovider", "test_clean_fees.py"], plugins=[OneLinePerTest()])
print("pytest exit status:", int(code))
sys.exit(int(code))

test_clean_fees.py::test_old_city_names_become_current_ones: passed
test_clean_fees.py::test_other_columns_are_untouched: passed
test_clean_fees.py::test_cleaned_file_matches_the_schema: failed
    Failed: 1 failure case(s): amount in_range(1, 200000): -4500
pytest exit status: 1

import pandas as pd
import pandas.testing as pdt
import pandera.pandas as pa
import pytest

from clean_fees import clean_fees
from fee_schema import fee_schema


def test_old_city_names_become_current_ones():
    raw = pd.DataFrame({"city": [" Bombay", "Cochin ", "Pune"]})
    expected = pd.DataFrame({"city": ["Mumbai", "Kochi", "Pune"]})
    pdt.assert_frame_equal(clean_fees(raw), expected)


def test_other_columns_are_untouched():
    raw = pd.read_csv("fees.csv", parse_dates=["paid_on"])
    out = clean_fees(raw)
    pdt.assert_frame_equal(out.drop(columns="city"), raw.drop(columns="city"))


def test_cleaned_file_matches_the_schema():
    out = clean_fees(pd.read_csv("fees.csv", parse_dates=["paid_on"]))
    try:
        fee_schema.validate(out, lazy=True)
    except pa.errors.SchemaErrors as err:
        found = [f"{c.column} {c.check}: {c.failure_case}" for c in err.failure_cases.itertuples()]
        pytest.fail(f"{len(found)} failure case(s): " + "; ".join(found))

"""A cleaning step of the fee pipeline: today's export writes some cities by their old names."""
OLD_NAMES = {"Bombay": "Mumbai", "Cochin": "Kochi"}


def clean_fees(raw):
    out = raw.copy()
    out["city"] = out["city"].str.strip().replace(OLD_NAMES)
    return out

The first two tests pass: clean_fees() maps old city names to current ones and leaves the other columns alone. The third test runs the schema on the cleaned file and fails, because cleaning fixed the city but not the refund row. That is the point of putting a schema into a test: the test suite now knows the contract too. In a project you would simply run pytest; the small runner prints one line per test instead, because pytest’s own summary also prints how long the run took, which changes every time.

Keep the two passes and real negative-payment failure rather than rewriting the original test result.

Choose the boundary and report needed for each rule

The same rule can live in three places, and each place has strengths:

Schema validation and hand-written checks express whole-table rules. Choose by required inputs, reporting and boundaries; constraints guard stored-data rules. The parts below cover what a first schema usually gets wrong: coercion converts types but does not clean text, and quarantine sets failing rows aside while the rest load. For tests, test fixturestest fixture: A deliberately selected input and expected behaviour used by a test to evaluate a stated code contract. are small inputs saved next to the tests, and property-based testing with Hypothesis generates inputs automatically and can shrink an observed failure to a simpler case under its strategy and settings.

Compare the required boundary and failure report across validation layers.

Schemas for real exports: three traps, parsing first, and a failure report Pro: opens with a Pro, Premium or Ultimate pass

Testing analysis code: fixture files, two versions and generated inputs Pro: opens with a Pro, Premium or Ultimate pass

Schema, hand-written check or database constraint: a measured choice Pro: opens with a Pro, Premium or Ultimate pass

Practise the original quality contracts

Interview questions

Key takeaways

• A pandera schema states each column’s dtype, checks and options once; strict=True also rejects unexpected columns, and columns are not nullable unless you say so.
• Eager validation stops at the first problem; lazy=True collects eligible failures in failure_cases, while parse and coercion failures can limit later checks.
• pytest finds test_ functions in test_ files; assert_frame_equal compares values, dtypes, columns and index.
• Tests check the code on inputs you chose; schemas check the data on every run. A schema inside a test brings the two together.