BadgeprintingEvent registration · badge automation

Validating Attendee Data with Pydantic Before Ingestion: Fixing Silent Coercion & Poisoned Batches

Symptom Statement

Badge fulfillment queues stall mid-batch with UnicodeDecodeError or a KeyError on a field the renderer assumed was always present. Reconciliation dashboards report persistent $0.00 discrepancies despite confirmed gateway authorizations, and async worker pools exhaust their retry budgets and crash with TypeError: unsupported operand type(s) for +: 'NoneType' and 'str'. These are not isolated bugs — they are cascading symptoms of malformed records slipping past the ingestion boundary, where downstream systems assume canonical shapes and silently coerce, drift, or die when the shape is wrong. This page is the attendee-payload contract documented under schema validation pipelines, the deterministic gate stage of the Registration Ingestion & Payment Reconciliation pipeline — it covers the exact Pydantic v2 model and fast-fail wrapper that turn untrusted external payloads into canonical records before they reach a broker, a ledger, or a badge template.

Root Cause Analysis

Three concrete faults produce this class of failure. Each sits at a different boundary — the producer contract, the Python type system, and the enqueue path — and each survives a naive “we already parse the JSON” defense.

  1. Divergent producer contracts. Registration data arrives from two upstreams with incompatible guarantees. Deterministic pull from form API polling returns paginated arrays with inconsistent casing (ticketPrice vs ticket_price), while event-driven payment webhook handling ships payloads whose schema version shifts under you. Neither stream promises a stable shape, and a renamed key becomes a missing field the moment a provider ships a release.
  2. Silent type coercion. Python’s dynamic typing masks bad values until execution reaches the template engine or the accounting job. A ticket_price of "29.95" sails through as a string and detonates decimal math downstream; a dietary_restrictions array carrying a null element corrupts CSV export; trailing whitespace in company_name breaks the badge alignment pass. The value looks fine at ingest and is fatal three stages later.
  3. No fast-fail gate before enqueue. With no hard contract at the trust boundary, both streams inject heterogeneous data straight into the message broker. One malformed record then poisons an entire chunk — the worker draining it dies, the chunk is redelivered, and the same bad record kills the next worker, amplifying a single dirty payload into a stalled queue.
Pydantic ingestion gate: canonical dict out, or categorized dead-letter Form polling and payment webhooks feed one stateless Pydantic gate; valid records emit a canonical dict to async batch processing, while a ValidationError is categorized and dead-lettered at a per-category log level (critical schema drift, error business-rule, warning syntax) with a payload fingerprint. Form Polling camelCase · paginated Payment Webhooks schema version drifts untrusted external payloads Pydantic Gate validate_and_route() strict · extra=forbid stateless gate one dict out — or None Async Batch Processing canonical records valid → canonical dict model_dump(mode="json") ValidationError · categorize_error() UNKNOWN_FIELD — schema drift logger.critical · page on-call BUSINESS_RULE_VIOLATION logger.error SYNTAX_ERROR — formatting logger.warning · recoverable Dead-Letter Queue tagged with payload_fp

Symptom-to-Resolution Matrix

Divergent Producer Contracts

Symptoms

  • The renderer raises KeyError: 'ticket_price' on payloads from one provider but not another.
  • A working ingest path breaks after an upstream provider ships a release, with no code change on your side.
  • Fields arrive camelCase from polling and snake_case from webhooks against the same model.

Root cause. No canonical field spec exists at the edge, so every producer’s casing and versioning quirks flow straight through.

Fix steps

  1. Pin one canonical snake_case schema and accept producer aliases explicitly with populate_by_name=True plus Field(alias=...), so ticketPrice and ticket_price both map to one field.
  2. Reject anything you did not declare with extra="forbid", converting silent schema drift into a loud, categorizable error at ingest.
  3. Disable implicit coercion with strict=True so a value that is the wrong type fails here rather than being quietly cast.
PYTHON
from pydantic import Field
# camelCase from polling and snake_case from webhooks both resolve here
ticket_price: Decimal = Field(alias="ticketPrice", ge=0, decimal_places=2)

Silent Type Coercion

Symptoms

  • Reconciliation shows $0.00 gaps even though the gateway confirms the charge.
  • CSV exports contain empty cells or a literal null where a diet tag should be.
  • Badge text is misaligned by a leading space no one can see in the data.

Root cause. Values enter as the wrong type or with junk whitespace, and dynamic typing defers the failure until arithmetic or rendering forces it.

Fix steps

  1. Sanitize strings before typing with a BeforeValidator attached through Annotated, stripping whitespace and null bytes at the door.
  2. Parse money into Decimal, never float — strip currency symbols in a mode="before" validator so "$29.95" and 29.95 both land as Decimal("29.95").
  3. Filter None out of list fields and deduplicate, so a null in dietary_restrictions can never reach the CSV writer.
PYTHON
def sanitize_string(v: str | None) -> str | None:
    if v is None:
        return None
    return v.strip().replace("\x00", "")  # kill whitespace + null bytes at ingest

No Fast-Fail Gate Before Enqueue

Symptoms

  • One bad record redelivers forever and stalls the whole chunk.
  • Workers crash-loop on TypeError while draining a batch that was 99% clean.
  • The dead-letter queue is empty even though records are clearly being lost.

Root cause. Validation happens (if at all) inside the worker after enqueue, so a poison payload is already in the broker before anything checks it.

Fix steps

  1. Validate with model_validate() before the record touches the broker — the gate returns a canonical dict or nothing.
  2. Categorize each ValidationError (syntax, business rule, schema drift) so recoverable and fatal failures route differently instead of all retrying blindly.
  3. Route failures to the dead-letter queue with a deterministic payload fingerprint so the exact upstream is traceable during triage.
PYTHON
# gate at the edge: canonical dict out, or None (never a half-valid model)
clean = validate_and_route(payload)
if clean is None:
    return  # already dead-lettered; never enqueued

Minimal Working Implementation

A single self-contained path: the canonical model with strict typing and business rules, a categorizing gate that emits a canonical dict or dead-letters the record, and a bounded async fan-out that keeps the gate off the broker’s hot path. The records this produces feed async batch processing; the reconciliation and windowing beyond this boundary are owned there, not here.

PYTHON
import asyncio
import hashlib
import logging
import re
from decimal import Decimal
from datetime import date
from enum import Enum
from typing import Annotated, Optional, Literal
from pydantic import (
    BaseModel, Field, field_validator, model_validator,
    BeforeValidator, ValidationError, ConfigDict,
)

logger = logging.getLogger("ingestion.validation")

EMAIL_RE = re.compile(r"^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$")


def sanitize_string(v: str | None) -> str | None:
    if v is None:
        return None
    return v.strip().replace("\x00", "")


# BeforeValidator MUST be carried via Annotated in Pydantic v2 — not class-level assignment
SanitizedStr = Annotated[str, BeforeValidator(sanitize_string)]
OptionalSanitizedStr = Annotated[Optional[str], BeforeValidator(sanitize_string)]


class AttendeeIngestionSchema(BaseModel):
    model_config = ConfigDict(
        strict=True,            # no silent coercion
        extra="forbid",         # unknown field == schema drift, caught here
        populate_by_name=True,  # accept both canonical name and producer alias
        validate_assignment=True,
    )

    attendee_id: str = Field(min_length=1, max_length=64)
    email: str
    first_name: SanitizedStr = Field(min_length=1, max_length=50)
    last_name: SanitizedStr = Field(min_length=1, max_length=50)
    ticket_type: Literal["standard", "vip", "press", "speaker"]
    ticket_price: Decimal = Field(alias="ticketPrice", ge=Decimal("0.00"), decimal_places=2)
    dietary_restrictions: list[str] = Field(default_factory=list)
    company_name: OptionalSanitizedStr = Field(default=None, max_length=100)
    registration_date: date

    @field_validator("email")
    @classmethod
    def validate_email_format(cls, v: str) -> str:
        if not EMAIL_RE.match(v):
            raise ValueError("Invalid email format")
        return v.lower()

    @field_validator("ticket_price", mode="before")
    @classmethod
    def coerce_price(cls, v):
        if isinstance(v, str):
            cleaned = re.sub(r"[^\d.]", "", v)  # strip currency symbols/whitespace
            if not cleaned:
                raise ValueError("Price string contains no numeric value")
            return Decimal(cleaned)
        if isinstance(v, (int, float)):
            return Decimal(str(v))  # str() first — never Decimal(float) directly
        raise ValueError("Unsupported price type")

    @field_validator("dietary_restrictions")
    @classmethod
    def filter_null_restrictions(cls, v: list) -> list[str]:
        seen: set[str] = set()
        out: list[str] = []
        for item in v:
            if item is None:
                continue
            s = str(item).strip()
            if s and s not in seen:
                seen.add(s)
                out.append(s)
        return out

    @model_validator(mode="after")
    def validate_business_rules(self) -> "AttendeeIngestionSchema":
        if self.ticket_type == "speaker" and self.ticket_price > Decimal("0.00"):
            raise ValueError("Speaker tickets must be complimentary (price=0.00)")
        if self.ticket_type == "vip" and not self.company_name:
            raise ValueError("VIP attendees require a valid company_name")
        return self


class ErrorCategory(str, Enum):
    SYNTAX_ERROR = "syntax_error"                    # recoverable formatting fault
    BUSINESS_RULE_VIOLATION = "business_rule_violation"
    UNKNOWN_FIELD = "unknown_field"                  # schema drift — page on it


def categorize_error(err: ValidationError) -> ErrorCategory:
    for e in err.errors():
        if e["type"] == "extra_forbidden":
            return ErrorCategory.UNKNOWN_FIELD
        if e["type"] in ("value_error", "assertion_error"):
            return ErrorCategory.BUSINESS_RULE_VIOLATION
    return ErrorCategory.SYNTAX_ERROR


def validate_and_route(payload: dict) -> dict | None:
    """Canonical dict out, or None after dead-lettering. Never a half-valid model."""
    try:
        return AttendeeIngestionSchema.model_validate(payload).model_dump(mode="json")
    except ValidationError as e:
        category = categorize_error(e)
        detail = {
            "category": category.value,
            "fields": [err["loc"] for err in e.errors()],
            "payload_fp": hashlib.sha256(
                str(sorted(payload.items())).encode()
            ).hexdigest()[:16],  # deterministic fingerprint for triage
        }
        if category == ErrorCategory.UNKNOWN_FIELD:
            logger.critical("schema_drift %s", detail)   # provider shipped a new field
        elif category == ErrorCategory.BUSINESS_RULE_VIOLATION:
            logger.error("business_rule %s", detail)
        else:
            logger.warning("syntax_error %s", detail)
        return None  # dead-letter; never enqueued


class IngestionPipeline:
    def __init__(self, max_concurrent: int = 50):
        self.semaphore = asyncio.Semaphore(max_concurrent)

    async def process_batch(self, payloads: list[dict]) -> list[dict]:
        async def _one(p: dict) -> dict | None:
            async with self.semaphore:
                # validation is CPU-bound — keep it off the event loop
                return await asyncio.to_thread(validate_and_route, p)

        results = await asyncio.gather(*[_one(p) for p in payloads])
        return [r for r in results if r is not None]  # drop the dead-lettered


if __name__ == "__main__":
    # Verification: dirty payload is rejected, clean payload survives with canonical types.
    dirty = {"attendee_id": "a1", "email": "[email protected]", "first_name": " Ada ",
             "last_name": "Byte", "ticket_type": "vip", "ticketPrice": "$0.00",
             "dietary_restrictions": ["vegan", None, "vegan"],
             "registration_date": "2026-07-02"}  # VIP with no company_name -> rejected
    assert validate_and_route(dirty) is None

    clean = {**dirty, "company_name": " Acme Corp ", "ticketPrice": "$120.00"}
    out = validate_and_route(clean)
    assert out["ticket_price"] == "120.00"          # Decimal, not float
    assert out["email"] == "[email protected]"                # lowercased
    assert out["dietary_restrictions"] == ["vegan"]  # null dropped, deduped
    assert out["company_name"] == "Acme Corp"        # whitespace stripped
    print("gate OK:", out)

Memory & Performance Constraints

Component Constraint Mitigation
Validation latency strict=True skips coercion but adds per-field checks Accept it at the edge; correctness outweighs raw throughput here
Model construction AttendeeIngestionSchema(**payload) pays __init__ overhead Use model_validate(payload) — it hits the compiled core directly
Regex cost Per-record re.compile inside a validator is wasted work Pre-compile patterns (EMAIL_RE) at module load
CPU vs event loop Validation is CPU-bound and blocks async I/O Run through asyncio.to_thread, bound by an asyncio.Semaphore
Serialization weight Pydantic model objects are heavier than dicts in the broker Emit model_dump(mode="json") before enqueue, never the model
Worker RSS on bursts Large batches hold every payload resident Chunk at batch_size≈200; cap max_concurrent at vCPU count

Incident Triage & Rollback

Target under 15 minutes to a clean fallback. Contain the poisoned stream first; never restart workers before the drift is patched, or the same record re-poisons the fresh process.

  1. Find the poison. Grep worker logs for the category and fingerprint: grep 'schema_drift' app.log | tail -20 — the payload_fp pins the exact upstream provider.
  2. Isolate the stream. Disable the failing ingestion route via feature flag (export VALIDATION_ROUTE_DISABLED=webhooks) so clean streams keep flowing while you patch.
  3. Inspect the dead-letter queue. redis-cli LLEN ingestion.dlq to size the blast radius, then sample one: redis-cli LINDEX ingestion.dlq 0.
  4. Replay cleaned payloads. Pull from the DLQ and re-run validate_and_route offline; patch the fixable formatting errors and re-enqueue only records that now return a dict.

Rollback to audit mode. If strict validation rejects legitimate payloads because of an unforeseen drift, do not disable validation outright — switch it to non-blocking: export VALIDATION_MODE=audit logs every ValidationError but lets the payload pass, so you capture the drift pattern without dropping records. Patch AttendeeIngestionSchema with the missing Optional field or alias, then re-run the gate against 10k historical payloads locally before redeploying.

Rollback validation. Redeploy with VALIDATION_MODE=enforce, watch the category counters for 15 minutes, and confirm the syntax_error rate holds below 0.5% and schema_drift returns to zero before restoring full ingestion load.

Frequently Asked Questions

Why strict=True if it rejects payloads that “would have worked” after coercion? Because the coercion is exactly the failure mode. A ticket_price silently cast from "29.95" looks fine at ingest and breaks decimal reconciliation three stages later, with no trace back to this boundary. Strict mode moves that failure to the one place that can categorize it and dead-letter the record — the edge — instead of a worker crash mid-batch. Where a producer legitimately sends the wrong type, you handle it explicitly in a mode="before" validator, not implicitly.

Why parse money into Decimal instead of float? float cannot represent most base-10 fractions exactly, so 0.1 + 0.2 is not 0.3 and a summed ledger drifts by cents that never reconcile. Decimal is exact for the two-place amounts tickets use. Note the code does Decimal(str(v)), never Decimal(float_value) directly — constructing a Decimal from a float inherits the float’s binary rounding error.

Should BeforeValidator be assigned as a class attribute like in some v1 examples? No. In Pydantic v2 a BeforeValidator must be carried on the field’s type via Annotated (Annotated[str, BeforeValidator(sanitize_string)]). A class-level assignment silently does nothing, so the sanitizer never runs and whitespace or null bytes reach your downstream unstripped.