Schema Validation Rules for FSMA 204 Supplier Data Ingestion

Regulatory compliance under FSMA 204 is not achieved at the point of recall; it is engineered at the boundary of data ingestion. When supplier payloads enter a traceability system, they must be validated against precise Key Data Element (KDE) schemas before they are committed to the lot graph. A single malformed timestamp, missing Traceability Lot Code, or non-standardized unit of measure fractures downstream Critical Tracking Event (CTE) lineage and invalidates automated recall workflows. Within the broader Supplier Data Ingestion & Sync Automation pipeline, this page owns one narrow but load-bearing responsibility: the strict, non-negotiable gate that every normalized record must pass before it reaches the traceability ledger.

The triggering requirement is concrete. FSMA 204 Subpart S obligates a regulated facility to produce sortable, electronic traceability records within 24 hours of an FDA request during an outbreak investigation. That guarantee is only credible if every persisted CTE carries a complete, well-typed KDE set. Schema validation is the control that makes the guarantee true: it converts fragmented, vendor-specific input into a uniform contract, rejects records that would silently corrupt the lot chain, and preserves the evidence an auditor needs to understand why a record was rejected. This page defines the validation architecture, the KDE data contract, a runnable pydantic v2 implementation with tenacity-guarded persistence, the quarantine strategy that keeps rejected records audit-visible, and the operational configuration to run the gate in production.

KDE Boundary Validation Architecture

Validation must execute immediately after payload normalization and strictly before database persistence. Regardless of whether the pipeline processes flat files, EDI 856/810 transactions, or RESTful JSON payloads, the transport layer should first emit a canonical dictionary. That separation of concerns keeps parsing logic isolated from compliance enforcement — a pattern reinforced by the CSV/EDI Parser Setup, which normalizes heterogeneous supplier files before this gate ever runs. Whether a record arrives as a flat-file row or a REST-polled object, it converges on the same schema check, so a CSV record and an API record become indistinguishable once they reach the ledger.

In high-throughput environments, the validation window must align with the ingestion cadence. If the system relies on scheduled fetches or webhook-driven streams, the API Polling Strategies should define explicit batch boundaries that enable synchronous schema evaluation without introducing upstream latency. Crucially, validation failures must never halt the entire pipeline. Invalid records are quarantined, logged, and routed for remediation while compliant payloads proceed to the ledger. This fail-forward design preserves availability while maintaining strict regulatory boundaries — one malformed record never aborts a supplier batch.

The gate is a three-stage funnel. Stage one checks structural completeness: are all required KDEs present for this CTE type? Stage two checks type and format: is the timestamp ISO 8601 with an explicit offset, is the quantity a positive number, is the unit of measure inside the controlled vocabulary? Stage three checks business rules: does the record preserve CTE continuity — for example, does a Shipping event carry a destination location, and does an event timestamp precede system-ingestion time? A record that clears all three stages is committed; a record that fails any stage is quarantined with its precise failure path intact.

Figure — KDE boundary validation gates:

Precise KDE Mapping Requirements

FSMA 204 mandates specific KDEs for each CTE type. The validation schema must enforce strict typing, format constraints, and business rules across the core elements below. This is the validation-scoped view of the complete field catalog documented in the KDE Field Mapping Guide, which covers every transport and CTE across the compliance model.

Optional fields such as reference_document_id or carrier_scac should pass through without blocking ingestion, provided they conform to expected data types. The engine must, however, reject partial KDE sets that break CTE continuity — for example, a Shipping event missing a destination location_id, or a Receiving event lacking a traceability_lot_code. Optional KDEs should default to an explicit null rather than an empty string, so downstream recall queries can distinguish “not applicable” from “missing and required.” Malformed records are never silently coerced; they are quarantined with full context for reconciliation.

Production Validation Engine

Implementing this in Python requires declarative schema enforcement, resilient error handling, and audit-ready logging. The model below uses pydantic v2 for validation and tenacity to guard the two I/O-bound handoffs — the durable quarantine write and the ledger publish — that can fail transiently. The validation itself is deterministic and is never retried; only the transport-class writes around it are. Structured logging captures the payload fingerprint, the CTE type, and the precise failure path for every decision.

from __future__ import annotations

import hashlib
import json
import logging
from datetime import datetime, timezone
from decimal import Decimal
from enum import Enum

from pydantic import BaseModel, ConfigDict, Field, ValidationError, field_validator
from tenacity import retry, retry_if_exception_type, stop_after_attempt, wait_exponential

# Structured audit logger; in production attach a JSON formatter and ship to SIEM.
audit_logger = logging.getLogger("fsma204.validation")
audit_logger.setLevel(logging.INFO)

UOM_ENUM = {"kg", "lb", "case", "pallet", "ea", "liter", "gallon"}


class CTEType(str, Enum):
    HARVESTING = "Harvesting"
    COOLING = "Cooling"
    INITIAL_PACKING = "Initial Packing"
    SHIPPING = "Shipping"
    RECEIVING = "Receiving"
    TRANSFORMATION = "Transformation"


class TransientWriteError(Exception):
    """Raised when a durable store rejects a write for a retryable reason."""


class KDEPayload(BaseModel):
    # strict=True blocks silent coercion; extra="allow" preserves optional KDEs.
    model_config = ConfigDict(strict=True, extra="allow")

    traceability_lot_code: str = Field(..., min_length=3, max_length=50)
    product_description: str = Field(..., min_length=2)
    quantity: Decimal = Field(..., gt=0)
    unit_of_measure: str
    location_id: str = Field(..., min_length=5)
    event_timestamp: str
    cte_type: CTEType

    @field_validator("traceability_lot_code")
    @classmethod
    def validate_lot_code(cls, v: str) -> str:
        if not v.replace("-", "").replace("_", "").isalnum():
            raise ValueError("Lot code must be alphanumeric (hyphen/underscore allowed)")
        return v

    @field_validator("unit_of_measure")
    @classmethod
    def validate_uom(cls, v: str) -> str:
        if v not in UOM_ENUM:
            raise ValueError(f"Unit of measure must be one of {sorted(UOM_ENUM)}")
        return v

    @field_validator("event_timestamp")
    @classmethod
    def validate_timestamp(cls, v: str) -> str:
        # Enforce ISO 8601 with an explicit offset, coerce to UTC, reject the future.
        try:
            dt = datetime.fromisoformat(v.replace("Z", "+00:00"))
        except ValueError as exc:
            raise ValueError("Timestamp must be valid ISO 8601") from exc
        if dt.tzinfo is None:
            raise ValueError("Timestamp must include a timezone offset")
        if dt.astimezone(timezone.utc) > datetime.now(timezone.utc):
            raise ValueError("Event timestamp cannot be in the future")
        return v


def _fingerprint(record: dict[str, object]) -> str:
    raw = json.dumps(record, sort_keys=True, separators=(",", ":"), default=str)
    return hashlib.sha256(raw.encode("utf-8")).hexdigest()[:12]


@retry(
    retry=retry_if_exception_type(TransientWriteError),
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=0.5, max=8),
    reraise=True,
)
def persist_quarantine(document: dict[str, object]) -> None:
    """Write one quarantine artifact to a durable, versioned store.

    Deterministic validation never enters this function; only the transient
    write path is retried, so a flaky object-store call does not lose evidence.
    """
    # store.put(document["payload_hash"], document)  # replace with real client
    audit_logger.warning(
        "KDE_REJECTED | hash=%s | errors=%s",
        document["payload_hash"],
        document["validation_errors"],
    )


def validate_supplier_batch(
    raw_records: list[dict[str, object]],
) -> tuple[list[dict[str, object]], list[dict[str, object]]]:
    valid_records: list[dict[str, object]] = []
    invalid_records: list[dict[str, object]] = []

    for idx, record in enumerate(raw_records):
        payload_hash = _fingerprint(record)
        try:
            validated = KDEPayload(**record)
        except ValidationError as exc:
            error_summary = [
                {"loc": err["loc"], "msg": err["msg"]} for err in exc.errors()
            ]
            document = {
                "original_payload": record,
                "validation_errors": error_summary,
                "payload_hash": payload_hash,
                "quarantined_at": datetime.now(timezone.utc).isoformat(),
            }
            persist_quarantine(document)  # tenacity-guarded durable write
            invalid_records.append(document)
            continue

        valid_records.append(validated.model_dump(mode="json"))
        audit_logger.info(
            "KDE_VALID | record_idx=%d | hash=%s | cte_type=%s",
            idx,
            payload_hash,
            validated.cte_type.value,
        )

    return valid_records, invalid_records

The model enforces the full contract declaratively: strict=True blocks pydantic’s implicit coercion so a string "12" is never silently accepted where a Decimal quantity is required, the controlled unit_of_measure enum rejects any unit outside the approved vocabulary, and the timestamp validator guarantees timezone awareness before coercing to UTC. Because validate_supplier_batch catches ValidationError per record and continues, one bad row is isolated without aborting the batch — the partial-commit contract that keeps ingestion available under real supplier data.

Error Handling and Quarantine Strategy

The engine draws a hard line between two failure classes and handles each differently. Validation faults — a lot code with illegal characters, a negative quantity, a timezone-naive timestamp, a unit outside the enum — are deterministic. Retrying them changes nothing, so they are routed straight to quarantine with zero validation retries. Transport faults — a durable-store write that times out, an object-store 503, a transient network partition on the quarantine or ledger handoff — are non-deterministic and often succeed on a second attempt, so persist_quarantine wraps only those in a tenacity retry with exponential backoff. Conflating the two either wastes cycles retrying a permanently malformed record or, worse, silently loses evidence when a transient write fails.

Quarantine is not merely error handling; it is a regulatory artifact. Each quarantined record is written as a self-contained JSON document carrying the original raw payload, the precise pydantic error path, the SHA-256 fingerprint, and a UTC timestamp. This preserves the full provenance an auditor needs to answer the only question that matters during a traceback: what did the supplier actually send, and why was it rejected? The engine processes records independently, so the valid records in a batch continue to the ledger while the bad record is isolated for reconciliation. This mirrors the dead-letter behavior of the shared Error Handling Workflows, so operators reconcile schema rejections through the same tooling as every other ingestion stage.

Quarantine depth per supplier is itself a signal. A sudden spike in rejections from one vendor usually means an upstream schema change — a renamed field, a switched date format, a new location-code convention — and that anomaly should surface to the Data Quality Monitoring layer against per-supplier SLAs rather than accumulate silently. When an entire supplier feed starts failing structural validation, the Fallback Routing Logic determines whether records divert to a manual review queue or halt for that vendor while healthy feeds continue.

Figure — two fault classes, two handling paths:

Integration with the Ingestion Pipeline

This gate is one enforcement point inside the parent Supplier Data Ingestion & Sync Automation pipeline, and it is deliberately narrow: its only job is to accept a normalized payload, decide, and hand off. Validated records flow into the message queue that fronts Async Batch Processing, where I/O-bound persistence and enrichment run on a worker pool decoupled from validation. For high-volume feeds, synchronous validate-and-write becomes a bottleneck; offloading heavy database writes to workers maintains backpressure and prevents memory exhaustion when a single batch carries hundreds of thousands of records.

The contract this gate enforces is identical to the one applied inside the CSV/EDI Parser Setup, so a flat-file record and a REST-polled record are indistinguishable once they reach the ledger. Because the engine reads normalized input and emits fingerprinted records into distinct stores, its output honors the access-control expectations described in the Security Boundaries for Trace Data guidance — raw supplier payloads, quarantine artifacts, and validated KDEs each live in their own controlled store. New vendors are wired to this gate through Supplier Onboarding Automation, which registers the per-supplier field mapping before the first live batch runs.

Operational Notes

Deploy the validation engine as a triggered job — an object-storage event handler, a Kubernetes Job, or a Celery task fired on batch arrival — rather than a long-lived loop, so a crashed run is restarted cleanly by the scheduler and the normalized input remains the single source of truth. Recommended runtime and dependency versions:

• Python 3.10+ (the code uses from __future__ import annotations, list[...] / dict[...] generics, and the X | Y union style).
• pydantic ≥ 2.5 — the v2 field_validator / model_dump API. Do not mix in the v1 validator decorator.
• tenacity ≥ 8.2 for wait_exponential and retry_if_exception_type.

Configuration should come from the environment, never from code. At minimum, provide a QUARANTINE_STORE target (use versioned, access-controlled object storage in production), a LEDGER_QUEUE endpoint for validated records, and the per-supplier column_mapping that the parser applies before this gate sees a record. Every validation decision must be logged with enough context to satisfy FDA inspection: the payload fingerprint, decision timestamp, specific KDE failures, and remediation status. Retention policies must align with FSMA 204 recordkeeping mandates — traceability data must remain accessible and searchable for at least two years, as detailed in the Data Retention Policies and the FDA FSMA 204 Traceability Rule. Keep each supplier’s mapping and the UOM enum under version control so a schema change is a reviewable diff, not a silent production surprise. For the flat-file-specific walkthrough of this gate, see Validating supplier CSV against KDE schemas.

Frequently Asked Questions

Related

• Validating supplier CSV against KDE schemas — the flat-file-specific walkthrough of this validation gate.
• CSV/EDI Parser Setup — normalizes heterogeneous supplier files into the canonical records this gate validates.
• API Polling Strategies — the REST ingestion vector whose batch boundaries the gate evaluates synchronously.
• Error Handling Workflows — dead-letter routing and operator reconciliation for quarantined records.
• KDE Field Mapping Guide — the full field catalog behind this validation-scoped contract.

Up: Supplier Data Ingestion & Sync Automation — this schema gate is the compliance enforcement point of the parent ingestion pipeline.