Validating grant payloads with Pydantic v2 in batch

Problem statement #

You have a run of several thousand grant payloads — a nightly sponsor export, a re-poll of an award portfolio, a re-processed spreadsheet — and you need to validate every one against your institutional schema in a single pass, collect a complete per-record error report instead of aborting on the first bad row, quarantine the rejects with enough detail to fix them, and hand only the clean records downstream for an idempotent write.

This how-to lives under Schema Validation Pipelines, one of the ingestion layers in Automated Ingestion & Data Sync Workflows. The parent guide establishes the contract this page implements at batch scale: validation is a policy-enforcement gate, not a data-correction step, and a non-conforming record is routed to quarantine with a structured manifest rather than coerced or silently dropped. Here the focus is narrow and mechanical — how to make Pydantic v2 validate a whole batch efficiently and how to turn its ValidationError output into a routable manifest.

Prerequisites #

Before running the batch validator, confirm this environment and policy configuration:

• Python 3.10+ for modern type hints (list[dict], X | None) and datetime.now(timezone.utc).
• Libraries: pydantic>=2.5. The batch path uses TypeAdapter, which is core Pydantic — no extra dependency. Install with pip install "pydantic>=2.5".
• Environment variables (never hard-code these, per Security Boundary Configuration):
  • QUARANTINE_DIR — a writable path where rejected-record manifests are persisted, e.g. /var/lib/grants/quarantine.
• Policy config: the institutional F&A (facilities & administrative) rate ceiling and the approved-category enum, kept in version control alongside your University Policy Mapping Frameworks. The raw batch itself is produced upstream by API Polling & Portal Integration or a parsed spreadsheet; this page assumes a list[dict] is already in memory.

Step-by-step implementation #

The batch flows through four stages: a single reusable model and TypeAdapter are built once; the batch is validated row-by-row so a ValidationError on one record cannot stop the run; the results are partitioned into a valid set and an invalid set; invalid records are written to quarantine as structured manifests while the valid set is handed to the downstream upsert.

Figure: one reusable adapter validates every row; the batch is partitioned so valid records upsert while rejects go to quarantine — the counts must reconcile.

Step 1 — Define the grant model once and build a reusable TypeAdapter #

The Pydantic model is the policy boundary: field constraints and validators encode the 2 CFR 200 cost rules, so a malformed record can never reach the write path. Build the model once at module load, then wrap list[GrantPayload] in a single TypeAdapter. Re-instantiating a model or adapter per record rebuilds Pydantic’s internal validator core each time and dominates the runtime of a large batch — construct it once and reuse it.

from datetime import datetime, timezone
from decimal import Decimal
from enum import StrEnum
from typing import Any

from pydantic import BaseModel, Field, TypeAdapter, field_validator, ConfigDict


class BudgetCategory(StrEnum):
    PERSONNEL = "personnel"
    EQUIPMENT = "equipment"
    TRAVEL = "travel"
    MATERIALS = "materials"


class GrantPayload(BaseModel):
    # forbid unknown keys so a renamed sponsor field fails loudly, not silently.
    model_config = ConfigDict(extra="forbid", str_strip_whitespace=True)

    award_id: str = Field(pattern=r"^[A-Z]{2}-\d{4}-\d{6}$")
    pi_orcid: str = Field(pattern=r"^\d{4}-\d{4}-\d{4}-\d{3}[0-9X]$")
    budget_category: BudgetCategory
    fiscal_year: int = Field(ge=2000, le=2100)
    total_cost: Decimal = Field(ge=0)
    direct_cost: Decimal = Field(ge=0)
    indirect_cost: Decimal = Field(ge=0)

    @field_validator("indirect_cost")
    @classmethod
    def _cap_indirect(cls, v: Decimal, info) -> Decimal:
        # 2 CFR 200.414: indirect cost may not exceed the negotiated F&A ceiling
        # (60% of total here). Enforced at the boundary so no over-cap row is written.
        total = info.data.get("total_cost", Decimal(0))
        if v > total * Decimal("0.60"):
            raise ValueError("indirect_cost exceeds 2 CFR 200.414 F&A ceiling")
        return v

    @field_validator("total_cost")
    @classmethod
    def _reconcile(cls, v: Decimal, info) -> Decimal:
        direct = info.data.get("direct_cost", Decimal(0))
        indirect = info.data.get("indirect_cost", Decimal(0))
        if abs(v - (direct + indirect)) > Decimal("0.01"):
            raise ValueError("total_cost != direct_cost + indirect_cost")
        return v


# Build ONCE at import; reuse for every batch. This is the key performance lever.
GRANT_BATCH_ADAPTER = TypeAdapter(list[GrantPayload])

Step 2 — Validate the batch, collecting a ValidationError per row #

The instinct is to call GRANT_BATCH_ADAPTER.validate_python(batch) and get a fully typed list back. That works, but it fails fast: the first bad record raises and you learn nothing about the other 4,999. For an ingestion run you want the opposite — a complete census of every rejection so a compliance officer can fix the source in one pass. Iterate and validate each record with the single reusable model, catching ValidationError per row and keeping the raw record and its normalized error list together.

from pydantic import ValidationError


def validate_batch(records: list[dict[str, Any]]) -> tuple[list[GrantPayload], list[dict[str, Any]]]:
    """Validate every record independently. Returns (valid_models, error_rows).

    Collect-all-errors, never fail-fast: one bad payload must not hide the
    other rejections or block the valid records from being written.
    """
    valid: list[GrantPayload] = []
    errors: list[dict[str, Any]] = []

    for index, record in enumerate(records):
        try:
            valid.append(GrantPayload.model_validate(record))
        except ValidationError as exc:
            # .errors() gives one structured entry per violation in the record —
            # loc (field path), msg, and type (the failing validator).
            errors.append({
                "index": index,
                "award_id": record.get("award_id"),
                "violations": [
                    {"loc": list(e["loc"]), "msg": e["msg"], "type": e["type"]}
                    for e in exc.errors()
                ],
                "payload": record,
            })
    return valid, errors

For very large runs where even the valid models will not fit comfortably in memory, chunk the input — validate 5,000 records, flush the valid slice to the downstream writer, flush the errors to quarantine, then move to the next chunk. The model and adapter stay resident; only one chunk of results is held at a time.

Step 3 — Partition and quarantine the rejects with structured manifests #

The valid and invalid lists are the partition. Persist each rejected record as its own manifest file — award identifier, the violation list, the raw payload, a UTC timestamp, and a content fingerprint so the same reject is not re-quarantined under a second name on a re-run. A quarantined record is a work item, not a lost record: it carries everything a human needs to correct the source and re-submit.

import hashlib
import json
import logging
from pathlib import Path

logger = logging.getLogger(__name__)


def quarantine_rejects(error_rows: list[dict[str, Any]], quarantine_dir: str) -> int:
    """Write one structured manifest per rejected record. Never overwrite blindly:
    the fingerprint makes re-quarantining the same reject idempotent."""
    out = Path(quarantine_dir)
    out.mkdir(parents=True, exist_ok=True)
    written = 0

    for row in error_rows:
        canonical = json.dumps(row["payload"], sort_keys=True, separators=(",", ":"))
        fingerprint = hashlib.sha256(canonical.encode("utf-8")).hexdigest()
        manifest = {
            "quarantined_at": datetime.now(timezone.utc).isoformat(),
            "content_hash": fingerprint,
            "award_id": row["award_id"],
            "violations": row["violations"],
            "payload": row["payload"],
        }
        target = out / f"{fingerprint}.json"
        if not target.exists():
            target.write_text(json.dumps(manifest, indent=2))
            written += 1
        logger.warning("Quarantined award=%s violations=%d",
                       row["award_id"], len(row["violations"]))
    return written

Step 4 — Hand the valid records downstream for an idempotent upsert #

The valid set leaves this module as typed GrantPayload objects and is written by the persistence layer. This page deliberately stops at the boundary; the commit itself — dialect on-conflict upsert versus hand-written SQL — is the subject of the sibling how-to, SQLAlchemy Upsert vs Raw SQL for Idempotent Grant Records. The driver below wires the three stages together and returns a reconciliation summary the run can assert on.

def run_batch(records: list[dict[str, Any]], quarantine_dir: str) -> dict[str, int]:
    valid, errors = validate_batch(records)
    quarantined = quarantine_rejects(errors, quarantine_dir)
    # upsert_valid_records(valid) — see the sibling how-to for the write path.
    summary = {
        "read": len(records),
        "valid": len(valid),
        "rejected": len(errors),
        "quarantined": quarantined,
    }
    # Count parity is the correctness invariant: nothing may vanish.
    assert summary["valid"] + summary["rejected"] == summary["read"]
    logger.info("Batch summary: %s", summary)
    return summary

Schema and field reference #

Fields the batch model enforces before any record is admitted. Widen the enum or ceiling in version-controlled policy config, not in code.

Verification #

Confirm a batch run behaved correctly before trusting its output:

1. Count parity. Assert valid + rejected == read. Any gap means a record was silently swallowed — a defect, never an accepted state.
2. Reproduce a rejection. Take one quarantined manifest, feed its payload back through GrantPayload.model_validate in a REPL, and confirm the raised ValidationError.errors() matches the stored violations list exactly.
3. Dry-run idempotency of quarantine. Run the batch twice against the same rejects. The quarantine directory must contain exactly one file per distinct rejected payload; the second pass writes zero new files because the fingerprints already exist.
4. Spot-check policy enforcement. Submit a record whose indirect_cost is 65% of total and confirm it lands in quarantine with a type of value_error on the indirect_cost path, proving the 2 CFR 200.414 ceiling is live.

Troubleshooting #

Three gotchas specific to batch Pydantic validation:

• Fail-fast instead of collect-all. Calling TypeAdapter.validate_python over the whole list[Model] raises on the first bad record and discards the rest of the report. Use it only when you genuinely want to reject the entire batch atomically. For an ingestion run, validate per-record and accumulate exc.errors() so every rejection is captured in one pass.
• Memory blowup on very large batches. Holding the raw list, the valid models, and the error rows simultaneously can exhaust memory on six-figure runs. Chunk the input (e.g. 5,000 records), flush valid and invalid slices after each chunk, and keep the model and TypeAdapter resident across chunks rather than rebuilding them — rebuilding the validator core per chunk is the second most common cause of a slow run.
• Cross-field rules see empty info.data on already-failed fields. A field_validator reads sibling values from info.data, but a field that already failed its own validation is absent. Default the lookups (info.data.get("total_cost", Decimal(0))) so the reconciliation check degrades to a clear secondary error instead of a KeyError that masks the real violation.

Frequently asked questions #

Related #

• Up to the parent topic: Schema Validation Pipelines
• SQLAlchemy Upsert vs Raw SQL for Idempotent Grant Records
• Grant Lifecycle Architecture Design
• API Polling & Portal Integration