Validating Clinical DataFrames with pandera

When a batch of parsed Observation rows lands in a pandas DataFrame on its way to the warehouse, you need a schema contract that rejects a malformed LOINC code, a non-UCUM unit, or an impossible vital sign before those rows poison downstream analytics — and you need to see every failing row in one pass, not fail on the first bad cell. This page is the runnable pandera companion to the clinical data quality validation frameworks guide: it turns the abstract rule catalog into an executable DataFrameSchema with column typing, Check objects, cross-field invariants, and lazy validation that collects the full failure set for batch triage.

pandera fits the tabular stage of clinical ETL — after FHIR JSON or HL7 v2 has been flattened into rows (see optimizing pandas for FHIR JSON parsing) and after per-field type coercion has run. The schema is the last deterministic gate: valid rows continue, failing rows fork to a quarantine frame carrying the reason they were rejected.

Clinical rule → pandera Check (quick reference)

Keep this lookup next to the schema definition. Each clinical constraint maps to a specific Check factory or Column argument; the right-hand column is what you actually write.

Implementation pattern: an Observation schema with lazy triage

The schema below models a flattened clinical Observation frame. Columns are typed and coerced; a regex Check constrains LOINC codes; isin checks pin the UCUM unit vocabulary and the FHIR Observation.status codes; an in_range check bounds a physiologic value; the numeric result is nullable because a text or panel-header observation legitimately carries no number. A schema-level (wide) Check enforces the cross-field invariant that a measurement window cannot end before it starts — an invariant that no single column can express because it spans two.

The distinction between a column Check and a dataframe-level Check is the pivot of the whole design. A column check receives one Series and reports failures per element; a wide check receives the entire DataFrame and returns a boolean Series (or a scalar), letting you compare effective_start against effective_end, assert that a value_quantity is present whenever status == "final", or enforce any relationship between fields. Both run under the same lazy pass, so a single validate call surfaces column and cross-field defects together.

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

# FHIR R4 Observation.status codes (http://hl7.org/fhir/observation-status)
OBS_STATUS = ["registered", "preliminary", "final", "amended",
              "corrected", "cancelled", "entered-in-error", "unknown"]

# A curated UCUM subset for the vitals/labs this pipeline accepts.
UCUM_UNITS = ["mg/dL", "mmol/L", "mm[Hg]", "/min", "Cel", "%", "g/dL", "10*3/uL"]

observation_schema = DataFrameSchema(
    columns={
        # Unique business key; str, coerced, never null.
        "observation_id": Column(str, nullable=False, unique=True, coerce=True),
        # LOINC codes: 1-5 digits, hyphen, single check digit (e.g. 8867-4).
        "loinc_code": Column(
            str,
            Check.str_matches(r"^\d{1,5}-\d$"),
            coerce=True,
        ),
        # Status must be a member of the FHIR value set.
        "status": Column(str, Check.isin(OBS_STATUS), coerce=True),
        # UCUM unit constrained to the accepted vocabulary.
        "unit": Column(str, Check.isin(UCUM_UNITS), coerce=True),
        # Numeric result: physiologically bounded, but nullable for text obs.
        "value_quantity": Column(
            "float64",
            Check.in_range(0, 1000, include_min=True, include_max=True),
            nullable=True,
            coerce=True,
        ),
        # Measurement window as timezone-aware timestamps.
        "effective_start": Column("datetime64[ns, UTC]", coerce=True),
        "effective_end": Column("datetime64[ns, UTC]", coerce=True),
    },
    # Wide check: runs against the whole frame, not one column.
    checks=Check(
        lambda df: df["effective_start"] <= df["effective_end"],
        error="effective_start must be <= effective_end",
    ),
    strict=True,   # reject unexpected columns
    coerce=True,   # coerce at the frame level too
)

coerce=True tells pandera to cast each column to its declared dtype before running checks, so a loinc_code read as an object array of mixed types is normalized to str first. strict=True fails the frame if an unexpected column sneaks in from an upstream mapping change — cheap insurance against silent schema drift.

The same contract expressed as a class-based DataFrameModel reads closer to a typed record and is easier to reuse across modules:

from pandera import DataFrameModel, Field
from pandera.typing import Series
import pandas as pd

class ObservationModel(DataFrameModel):
    observation_id: Series[str] = Field(unique=True, nullable=False, coerce=True)
    loinc_code: Series[str] = Field(str_matches=r"^\d{1,5}-\d$", coerce=True)
    status: Series[str] = Field(isin=OBS_STATUS, coerce=True)
    unit: Series[str] = Field(isin=UCUM_UNITS, coerce=True)
    value_quantity: Series[float] = Field(
        in_range={"min_value": 0, "max_value": 1000}, nullable=True, coerce=True
    )
    effective_start: Series[pd.DatetimeTZDtype] = Field(dtype_kwargs={"unit": "ns", "tz": "UTC"})
    effective_end: Series[pd.DatetimeTZDtype] = Field(dtype_kwargs={"unit": "ns", "tz": "UTC"})

    class Config:
        strict = True
        coerce = True

    @pa.dataframe_check
    def window_ordered(cls, df: pd.DataFrame) -> Series[bool]:
        return df["effective_start"] <= df["effective_end"]

Both styles compile to the same validation engine — pick DataFrameSchema for schemas assembled dynamically, DataFrameModel when you want a named, importable type. Now the part that matters for batch clinical work: validate lazily so a single bad row does not hide the other ninety-nine.

import pandas as pd
import pandera as pa

def validate_observations(df: pd.DataFrame):
    """Validate a batch, routing failing rows to a quarantine frame.

    Returns (valid_df, quarantine_df). Lazy validation collects EVERY
    violation instead of raising on the first, which is essential when
    triaging a batch rather than a single record.
    """
    try:
        valid = observation_schema.validate(df, lazy=True)
        return valid, df.iloc[0:0].copy()   # empty quarantine
    except pa.errors.SchemaErrors as err:
        # failure_cases is a DataFrame: columns include
        # schema_context, column, check, failure_case, index.
        fc = err.failure_cases
        # Row-level failures carry the original index; wide/dataframe
        # checks may report index as NaN, so drop those before select.
        bad_index = fc["index"].dropna().unique()
        quarantine = df.loc[df.index.isin(bad_index)].copy()
        # Attach a PHI-safe reason: column + check only, never the value.
        reasons = (
            fc.dropna(subset=["index"])
              .groupby("index")
              .apply(lambda g: "; ".join(f"{c}:{k}" for c, k
                                         in zip(g["column"], g["check"])))
        )
        quarantine["quarantine_reason"] = quarantine.index.map(reasons)
        valid = df.loc[~df.index.isin(bad_index)].copy()
        return valid, quarantine

SchemaErrors (plural — the lazy variant) exposes failure_cases, a tidy DataFrame with one row per violated cell: schema_context (whether the failure came from a Column or a DataFrameSchema-level check), the column, the check that fired, the offending failure_case value, and the original index. That index is the join key back to the source frame, which is what makes clean row-level routing possible. Column checks always populate index; a dataframe-level wide check reports at the frame scope and can leave index null, which is exactly why the routing code calls dropna(subset=["index"]) before grouping — a naive groupby("index") would otherwise silently drop those violations or raise. When a wide check fails, treat the whole batch as suspect and reconcile the window-ordering defect separately rather than trying to attribute it to one row.

The quarantine_reason deliberately records only the column name and check identity (for example unit:isin or loinc_code:str_matches), never failure_case, so a quarantine report of a Patient.name-adjacent value cannot leak the value itself. Downstream operators triaging the quarantine frame get enough to diagnose the rule that broke without ever seeing protected data.

Validation and testing

Test the schema like any deterministic gate: a known-good frame must pass untouched, and a frame with one bad UCUM unit must land exactly that row in quarantine while the rest survive.

import pandas as pd
import pytest

def _frame(unit="mg/dL"):
    return pd.DataFrame({
        "observation_id": ["obs-1", "obs-2"],
        "loinc_code": ["2339-0", "8867-4"],   # glucose, heart rate
        "status": ["final", "final"],
        "unit": ["mg/dL", unit],
        "value_quantity": [95.0, 72.0],
        "effective_start": pd.to_datetime(["2026-01-01T09:00Z", "2026-01-01T09:05Z"], utc=True),
        "effective_end":   pd.to_datetime(["2026-01-01T09:00Z", "2026-01-01T09:05Z"], utc=True),
    })

def test_valid_frame_passes():
    valid, quarantine = validate_observations(_frame())
    assert len(valid) == 2
    assert quarantine.empty

def test_bad_unit_row_is_quarantined():
    valid, quarantine = validate_observations(_frame(unit="beats/min"))  # not UCUM
    assert len(valid) == 1
    assert list(quarantine["observation_id"]) == ["obs-2"]
    assert "unit" in quarantine.iloc[0]["quarantine_reason"]

test_bad_unit_row_is_quarantined proves the two behaviors that matter: the invalid unit row is isolated by its original index, and the healthy row still flows through. Add a companion case that flips effective_start past effective_end to confirm the wide check fires and the batch is flagged, and one that supplies a malformed LOINC like "999999-0" to confirm the regex rejects six-digit codes. For a fast pre-merge check outside pytest, call observation_schema.validate(df, lazy=True) in a notebook and inspect err.failure_cases.groupby("check").size() to see the failure distribution across a real batch — the single most useful triage view when an upstream feed regresses, since it ranks the broken rules by how many rows each one rejected.

Gotchas and compliance constraints

• Eager validation hides later errors. The default validate(df) raises SchemaError (singular) on the first failing check and stops, so a 50,000-row batch reports one problem when it may have thousands. Always pass lazy=True for batch clinical data and read SchemaErrors.failure_cases — otherwise every re-run only surfaces the next single defect, turning triage into a serial guessing game.

• coerce=True can mask real type problems. Coercion is convenient, but casting a garbage value_quantity like "N/A" to float64 yields NaN, which a nullable=True column then accepts silently. Decide per column whether missing-after-coercion is legitimately null or a parse failure; where it is a failure, keep the column non-nullable or add an explicit Check that the raw value was numeric before coercion ran.

• Large-frame check performance. Element-wise Python lambdas and heavy regex checks scale poorly on multi-million-row frames. Prefer vectorized built-ins (Check.isin, Check.in_range, Check.str_matches compile to pandas/numpy ops), validate in chunks, and reserve element_wise=True for rules that genuinely cannot be vectorized.

• Quarantined rows carry PHI. A failure_cases dump can echo a real value_quantity, a loinc_code, or an id tied to a patient. Reports and logs derived from quarantine must be PHI-safe: record the column and check identity, hash or redact the raw failure_case, and store the quarantine frame under the same access controls and retention rules as the source clinical data — never in plaintext observability tooling.

• Lazy by default — every batch validation call passes lazy=True and inspects failure_cases.

• Coercion audited — each coerce=True column has a documented decision on post-coercion nulls.

• PHI-safe quarantine — reports carry column + check only; raw values are redacted or hashed.

• Vocabularies pinned — UCUM unit set and Observation.status codes are version-controlled, not inline literals scattered across modules.

Related

• Clinical data quality validation frameworks — parent guide: where schema validation sits among clinical DQ controls.
• Type coercion for clinical data types — the coercion stage that runs before a pandera schema and interacts with coerce=True.
• Optimizing pandas for FHIR JSON parsing — how the flattened Observation frame is built upstream of validation.