data-validation-contracts

Data Validation and Contracts

Overview

Code that trusts its input computes wrong numbers quietly. A contract states what the data must look like at a boundary and stops the pipeline, with a readable message, when it does not. EDA finds the facts (eda-and-data-profiling); contracts keep them true.

Core principle: validate at every boundary, fail loudly with the column and the bad values, and never catch a validation error just to keep going.

1. What to check — the dimensions

Dimension	Question	Typical check
Completeness	Are required values present?	not-null on keys and required fields
Uniqueness	Is the grain what we think?	unique on the (composite) key
Validity	Are values in range and in the allowed set?	ge/le, isin, regex, dtype
Consistency	Do columns agree with each other?	shipped_at not null when status == "shipped"
Referential integrity	Do foreign keys point at something?	relationship test, anti-join count
Timeliness	Is the data fresh and the period complete?	max timestamp within N hours
Volume	Is the batch plausibly sized?	row count within a range of recent batches

Check the columns your logic depends on, not every column — a contract nobody can maintain gets disabled.

2. Which tool where — follow the repo

Boundary	Tool (if the repo has none yet)
DataFrames inside Python code and tests	pandera (import pandera.pandas as pa; pandera.polars for Polars)
Pipeline checkpoints with stored results and docs	Great Expectations (GX Core 1.x) — only if the repo already runs it
Single records: API payloads, configs, messages	Pydantic 2 models
Warehouse tables built by dbt	dbt data tests and model contracts (sql-analytics-and-dbt)

Never add a second validation library beside the one the repository uses (rule stack-choice).

3. pandera schemas as code

import pandas as pd
import pandera.pandas as pa
from pandera.typing import DataFrame, Series

class Orders(pa.DataFrameModel):
    order_id: Series[str] = pa.Field(unique=True)
    customer_id: Series[str]
    amount: Series[float] = pa.Field(ge=0)
    status: Series[str] = pa.Field(isin=["placed", "shipped", "refunded"])
    shipped_at: Series[pd.Timestamp] = pa.Field(nullable=True)

    class Config:
        strict = True
        coerce = True

    @pa.dataframe_check
    def shipped_orders_have_a_date(cls, df: pd.DataFrame) -> pd.Series:
        return ~((df["status"] == "shipped") & df["shipped_at"].isna())

@pa.check_types(lazy=True)
def clean_orders(raw: DataFrame[Orders]) -> DataFrame[Orders]:
    ...