Pydantic Schema Validation for Payments: The Ingestion Validation Gate #

Payment reconciliation fails quietly when a structurally plausible record carries a semantically wrong field. A routing number that fails its check digit, an amount decoded without its implied decimals, an out-of-range SEC code, or an effective date in the wrong timezone will all pass a fixed-width framer and still corrupt every downstream stage that trusts them. Schema validation is the hard gate that stops that contamination, and it sits within the broader Automated File Ingestion & Parsing Pipelines framework as the last transformation before records enter the matching engine. What breaks without it is not a single file — it is the invariant that every record reaching reconciliation is canonical, typed, and provably conformant.

This guide covers the Pydantic v2 patterns that make that gate deterministic. Records arrive already decoded against the byte-level NACHA record layouts by the upstream fixed-width file decoding stage, and Pydantic replaces the scattered if/else checks that would otherwise validate them with a single declarative model. Failures become structured ValidationError payloads that map to NACHA return codes and Reg E error-resolution obligations rather than silent drops or pipeline aborts. For the nested parent/child case — entry-detail records that carry their own remittance addenda — the validating NACHA addenda records with Pydantic guide extends the same model composition to Type 7 records.

Concept Definition: What the Validation Gate Guarantees #

A validation gate is a pure function from a raw decoded record to one of two outcomes: a frozen, fully typed model instance, or a structured error object. It never mutates shared state, never partially accepts a record, and never raises an uncaught exception past its own boundary. In Pydantic v2 terms, the gate is a BaseModel subclass whose ConfigDict pins the coercion policy and whose field- and model-level validators encode the payment-domain rules that byte-offset decoding cannot express.

The field-level contract for a NACHA Entry Detail (Type 6) record is precise. The record is exactly 94 bytes; after decoding, the validation model enforces the meaning of each field rather than its position:

• record_type — must equal 6. Any other value means the framer handed the wrong record type to this model.
• transaction_code — two digits; 21–29 are credits, 31–39 are debits. 22 (checking credit) and 27 (checking debit) dominate consumer volume; prenotes use 23/28.
• routing_number — nine ASCII digits whose ninth digit is a modulus-10 check digit over the first eight, weighted 3,7,1,3,7,1,3,7,1. A stripped leading zero (from an upstream float coercion) fails this check immediately.
• amount — ten digits with implied two-decimal scale. 0000012345 is $123.45, never $12,345. The model stores integer cents and exposes a decimal.Decimal accessor so no code path ever touches the value as a float.
• sec_code — a three-character Standard Entry Class code constrained to the NACHA allowlist (PPD, CCD, WEB, TEL, and the rest). An unknown code is a conformance failure, not a warning.
• trace_number — fifteen digits: the ODFI routing prefix (8) plus a sequence (7). It is unique within a file and is the primary key the matching engine keys on.

The gate's cost is linear in field count. For a batch of $n$ records with $f$ fields each, validation is $O(n\cdot f)$ with a constant per-field factor; because Pydantic v2's core is compiled in Rust, that constant is small enough that validation is almost never the throughput bottleneck — memory management around the batch is.

Architecture: Where Validation Sits in the Pipeline #

Validation is the seam between decoding and matching, and its position is deliberate. Bytes cross the secure transfer boundary, are framed and sliced by the positional decoder, and only then reach this gate. Records that validate are guaranteed canonical and flow forward to the transaction matching and reconciliation algorithms; records that fail are quarantined with a diagnostic payload and never touch the ledger. The gate is also where idempotency metadata — the source file hash and byte offset — is stamped onto each record so any later question can be answered from the original artifact.

Because reconciliation routinely processes multi-gigabyte settlement files, the gate must hold constant memory. It is expressed as a generator that consumes an iterator of decoded dictionaries and yields validated models or error payloads one at a time, exactly the streaming discipline the sibling high-volume Pandas parsing strategies guide applies to columnar workflows, and the concurrency model the async batch processing architectures guide uses to keep CPU-bound validation off the I/O event loop.

Phase-by-Phase Implementation #

The gate is built in four ordered steps: define the model and its coercion policy, encode field- and cross-field rules as validators, translate failures into compliance payloads, and drive the whole thing from a memory-bounded generator. Each step is independently testable.

Step 1 — Define the model and pin the coercion policy #

The ConfigDict is the single most important design decision. extra="forbid" rejects unexpected fields (a decoder bug that adds a stray key must fail loudly, not pass silently). frozen=True makes each validated record immutable and hashable, which is what lets the matching engine dedupe and route records across threads without defensive copies. Monetary amounts stay in integer cents and surface as Decimal through a computed accessor so float never enters the arithmetic path.

from __future__ import annotations

from datetime import date
from decimal import Decimal
from typing import Optional

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


class ACHEntryDetail(BaseModel):
    """Validated NACHA Entry Detail (Type 6) record. Immutable post-validation."""

    model_config = ConfigDict(
        strict=False,          # allow "6" -> 6 coercion from decoded strings
        extra="forbid",        # a stray field is a decoder bug, not a warning
        populate_by_name=True,
        frozen=True,           # thread-safe, hashable, dedupe-friendly
    )

    record_type: int = Field(ge=6, le=6, description="NACHA Record Type 6")
    transaction_code: int = Field(description="Standard Entry Detail transaction code")
    routing_number: str = Field(pattern=r"^\d{9}$", description="ABA routing/transit number")
    account_number: str = Field(min_length=1, max_length=17, description="DFI account number")
    amount_cents: int = Field(ge=0, le=9_999_999_999, description="Amount in cents (implied 2 dp)")
    individual_id: str = Field(max_length=15, description="Individual name or ID")
    addenda_indicator: int = Field(ge=0, le=1, description="1 if addenda present, else 0")
    trace_number: str = Field(pattern=r"^\d{15}$", description="ODFI routing (8) + sequence (7)")
    effective_date: Optional[date] = None
    sec_code: Optional[str] = Field(default=None, max_length=3)
    raw_record: Optional[str] = Field(default=None, description="Original bytes for the audit trail")

    @property
    def amount(self) -> Decimal:
        """Money as Decimal dollars — never float. `1234` cents -> Decimal('12.34')."""
        return (Decimal(self.amount_cents) / Decimal(100)).quantize(Decimal("0.01"))

Step 2 — Encode field and cross-field rules as validators #

field_validator handles single-field business logic; model_validator(mode="after") handles constraints that span fields. Keeping the raw record available means every rejection can be reproduced byte-for-byte during an audit.

    @field_validator("routing_number")
    @classmethod
    def validate_aba_checksum(cls, v: str) -> str:
        """ABA modulus-10 check: weighted 3,7,1 sum must be divisible by 10."""
        if len(v) != 9 or not v.isdigit():
            raise ValueError("Routing number must be exactly 9 digits.")
        weights = (3, 7, 1, 3, 7, 1, 3, 7, 1)
        if sum(int(d) * w for d, w in zip(v, weights)) % 10 != 0:
            raise ValueError(f"Invalid ABA checksum for routing number: {v}")
        return v

    @field_validator("transaction_code")
    @classmethod
    def validate_transaction_code(cls, v: int) -> int:
        """Defined ACH entry-detail codes: 21-29 (credits), 31-39 (debits)."""
        valid_codes = set(range(21, 30)) | set(range(31, 40))
        if v not in valid_codes:
            raise ValueError(f"Transaction code {v} is not a defined ACH entry-detail code.")
        return v

    @field_validator("sec_code")
    @classmethod
    def enforce_sec_allowlist(cls, v: Optional[str]) -> Optional[str]:
        if v is None:
            return v
        allowed = {"PPD", "CCD", "CTX", "WEB", "TEL", "ARC", "BOC", "POP", "RCK", "SHR", "IAT"}
        if v.upper() not in allowed:
            raise ValueError(f"Unsupported SEC code: {v!r}")
        return v.upper()

    @model_validator(mode="after")
    def enforce_addenda_trace_consistency(self) -> "ACHEntryDetail":
        """
        The trace-number prefix carries the ODFI id from the Batch Header, not this
        record's RDFI routing number, so a prefix match against routing_number is
        incorrect and belongs at the batch level. Here we only assert internal
        consistency: an addenda flag requires a trace number to link the child record.
        """
        if self.addenda_indicator == 1 and not self.trace_number:
            raise ValueError("Addenda flag set but trace number is missing.")
        return self

Step 3 — Translate failures into compliance payloads #

A raw ValidationError is not audit-ready. Operations teams query exceptions by return code and failed field, so the gate normalizes every error into a structured payload keyed to NACHA return codes. Use hashlib.sha256 for the record fingerprint — Python's built-in hash() is salted per process and is not stable across runs, which makes it useless for an audit trail.

import hashlib
from datetime import datetime, timezone
from typing import Any

from pydantic import ValidationError


def route_validation_exception(exc: ValidationError, raw_record: str) -> dict[str, Any]:
    """Transform a Pydantic ValidationError into an audit-ready compliance payload."""
    errors = exc.errors()
    mapped_codes: list[str] = []

    for err in errors:
        loc = ".".join(str(part) for part in err["loc"])
        msg = err["msg"].lower()

        if "routing_number" in loc and "checksum" in msg:
            mapped_codes.append("R13")   # invalid ACH routing number
        elif "account_number" in loc:
            mapped_codes.append("R03")   # no account / unable to locate account
        elif "amount" in loc:
            mapped_codes.append("R10")   # customer advises not authorized
        elif "sec_code" in loc:
            mapped_codes.append("R05")   # unauthorized debit to consumer account
        else:
            mapped_codes.append("R99")   # internal-use: unmapped schema failure

    return {
        "exception_type": "SCHEMA_VALIDATION_FAILURE",
        "nacha_return_codes": sorted(set(mapped_codes)),
        "failed_fields": [list(err["loc"]) for err in errors],
        "raw_record_sha256": hashlib.sha256(raw_record.encode("utf-8")).hexdigest(),
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "compliance_tier": "REG_E_AUDITABLE",
    }

Step 4 — Drive the gate from a memory-bounded generator #

Loading a multi-million-record file into a list to validate it will trigger an OOM kill during a peak settlement window. The gate is a generator: it yields a (model, None) pair for each pass and a (None, payload) pair for each failure, so the caller decides how to fan the two streams out without the gate ever holding more than one record.

from collections.abc import Iterator


def stream_validate_entries(
    raw_dicts: Iterator[dict[str, Any]],
) -> Iterator[tuple[Optional[ACHEntryDetail], Optional[dict[str, Any]]]]:
    """Validate decoded records one at a time; never materialize the whole batch."""
    for record in raw_dicts:
        raw = record.get("raw_record") or repr(record)
        try:
            yield ACHEntryDetail(**record), None
        except ValidationError as exc:
            yield None, route_validation_exception(exc, raw)

The caller consumes the two-stream shape directly, routing clean records to matching and failures to the exception queue:

def partition_batch(raw_dicts: Iterator[dict[str, Any]]) -> tuple[int, int]:
    """Return (validated_count, rejected_count) while streaming to sinks."""
    ok = bad = 0
    for entry, error in stream_validate_entries(raw_dicts):
        if entry is not None:
            forward_to_matching(entry)   # canonical, frozen, dedupe-safe
            ok += 1
        else:
            enqueue_exception(error)      # structured, regulator-ready
            bad += 1
    return ok, bad

Edge Cases & Known Failure Modes #

Schema validation removes a whole class of silent bugs, but a handful of Pydantic-specific and payment-specific traps still cause wrong-but-accepted records. The table below is the production checklist; each row is a real failure that has reached a ledger.

Compliance & Auditability #

The validation gate is where several regulatory obligations become concrete implementation rules, not abstractions.

• Reg E error resolution — 12 CFR 1005.11. Consumer error-resolution and provisional-credit clocks start from settlement events the pipeline must timestamp accurately. Preserving effective_date and the raw_record bytes for the full dispute window means a validated record can always be reconstructed for an investigation, and the REG_E_AUDITABLE tier on each exception payload marks records that fall under those timelines.
• NACHA Operating Rules — return-reason fidelity. Mapping validation failures to real return codes (R03, R05, R13, and the rest) keeps the exception ledger aligned with the codes an RDFI would actually transmit. The authoritative catalog is the NACHA Operating Rules; the SEC-code allowlist and the transaction-code ranges in the model are drawn directly from them, so a rules change is a one-line edit to a validator rather than a hunt through business logic.
• BSA/AML examination readiness. Every rejection carries a deterministic raw_record_sha256, a UTC timestamp, and the exact failed fields, so an examiner can verify that no record was dropped, altered, or double-processed. The SHA-256 fingerprint is stable across processes and runs, which is precisely why the salted built-in hash() is unacceptable here.

The design rule that ties these together: validation never fails silently. Malformed records are routed, not discarded; the raw bytes are retained alongside the canonical model; and the exception payload is structured so the audit answer is a query, not a log grep.

Testing & Verification #

The gate is verified against known-good and deliberately corrupt records with pytest. Construct fixtures whose routing numbers satisfy the ABA checksum so the happy path exercises the real validators rather than tripping on the fixture.

import pytest
from decimal import Decimal
from pydantic import ValidationError


def valid_entry(**overrides) -> dict:
    base = {
        "record_type": 6,
        "transaction_code": 22,          # checking credit
        "routing_number": "021000021",   # valid ABA checksum
        "account_number": "12345678901234567",
        "amount_cents": 12345,           # $123.45
        "individual_id": "ID-0000000001",
        "addenda_indicator": 0,
        "trace_number": "021000020000001",
        "sec_code": "PPD",
    }
    base.update(overrides)
    return base


def test_amount_is_decimal_with_implied_scale():
    entry = ACHEntryDetail(**valid_entry())
    assert entry.amount == Decimal("123.45")
    assert isinstance(entry.amount, Decimal)


def test_bad_aba_checksum_is_rejected():
    with pytest.raises(ValidationError) as exc:
        ACHEntryDetail(**valid_entry(routing_number="021000029"))
    assert "checksum" in str(exc.value).lower()


def test_unknown_sec_code_is_rejected():
    with pytest.raises(ValidationError):
        ACHEntryDetail(**valid_entry(sec_code="ZZZ"))


def test_exception_payload_maps_to_return_code():
    try:
        ACHEntryDetail(**valid_entry(routing_number="021000029"))
    except ValidationError as exc:
        payload = route_validation_exception(exc, raw_record="RAW-BYTES")
        assert "R13" in payload["nacha_return_codes"]
        assert payload["compliance_tier"] == "REG_E_AUDITABLE"
        assert len(payload["raw_record_sha256"]) == 64

A rejected record serializes to a stable JSON shape the exception ledger and any SIEM ingestion can assert against:

{
  "exception_type": "SCHEMA_VALIDATION_FAILURE",
  "nacha_return_codes": ["R13"],
  "failed_fields": [["routing_number"]],
  "raw_record_sha256": "9f2c0b1e7c4a3d5b6e8f0a1c2d3e4f5061728394a5b6c7d8e9f0a1b2c3d4e5f6",
  "timestamp": "2026-07-02T14:03:11.482913+00:00",
  "compliance_tier": "REG_E_AUDITABLE"
}

Frequently Asked Questions #

Related guides in this collection #

• Fixed-Width File Decoding — the positional decode stage that produces the records this gate validates.
• Validating NACHA Addenda Records with Pydantic — nested Type 7 model composition and cross-record checks for remittance and return entries.
• High-Volume Pandas Parsing Strategies — chunked, memory-bounded ingestion for delimited exports that feed the same validation gate.
• Async Batch Processing Architectures — keeping CPU-bound validation off the I/O event loop and routing failures to dead-letter queues.
• NACHA Record Layouts Explained — the byte-level field catalog behind every rule in this model.
• Automated File Ingestion & Parsing Pipelines — the parent guide mapping the full ingestion layer this gate closes.