How to Map FSMA 204 KDEs to SQL Schemas Without Type Coercion Failures

The most frequent production failure when mapping FSMA 204 Key Data Elements (KDEs) to relational SQL schemas is not missing data — it is silent type coercion and schema drift during heterogeneous payload ingestion. When Critical Tracking Event (CTE) records arrive from disparate ERP, WMS, or IoT endpoints, inconsistent timestamp precision, unstructured JSON expansion, and null-versus-mandatory field combinations routinely raise psycopg2.errors.InvalidDatetimeFormat or JSONB constraint violations. Worse, some payloads pass the database’s own type check but persist a wrong-but-valid-looking record — an empty-string lot_code that slips past a NOT NULL column, or a naive timestamp reinterpreted in the server’s local zone. Both outcomes break a lot chain you cannot reconstruct on demand. This guide resolves the failure with a deterministic validation boundary that enforces strict KDE typing before SQL insertion and routes non-compliant payloads to quarantine instead of dropping them.

Root Cause: Why Type Coercion Breaks KDE-to-SQL Mapping

FSMA 204 (21 CFR Part 1, Subpart S) treats each CTE as an independent obligation with a fixed set of mandatory KDEs. The rule specifies what every event must carry, but it says nothing about the shape of the data your upstream systems actually emit. That gap is where mapping fails. Three distinct root causes converge at the SQL boundary:

• Timestamp precision mismatch. A supplier sends 2024-03-15T08:30:00Z while the column is declared TIMESTAMP WITHOUT TIME ZONE, or a naive string like 2024-03-15 08:30:00 lands in a TIMESTAMPTZ column and is silently reinterpreted in the session timezone. FSMA 204 requires the date and time of an event; a coerced offset corrupts event sequencing and one-up/one-back ordering.
• JSONB expansion drift. Nested arrays in transformer_info or shipping_details exceed a column limit or violate a CHECK constraint. Because the raw structure is unbounded upstream, the schema drifts the moment a new trading partner adds a nested field your JSONB constraint never anticipated.
• Null versus mandatory KDEs. Fields like lot_code or traceability_lot_code arrive as empty strings rather than NULL. They bypass a NOT NULL constraint but fail downstream compliance checks, producing a persisted CTE that looks complete and is legally indefensible.

The underlying mistake is architectural: relying on the database engine to infer and enforce KDE typing. Database drivers rarely surface the precise field that failed — they return a generic constraint violation with a sqlstate code. The canonical field-level contract these payloads must satisfy is defined in the parent KDE Field Mapping Guide; the job here is to enforce that contract in application code, at the edge, before a single field reaches the SQL engine.

The following entity relationships define the target schema the mapping writes into — the append-only CTE table, its KDE and lot references, and the quarantine sink for rejected payloads:

The Canonical KDE-to-Column Contract

Every parser in the program must map its heterogeneous inputs onto exactly one column contract. The table below fixes the target SQL column, its type, the validation rule applied at the boundary, and the Subpart S provision that mandates the underlying KDE.

Persisting the raw payload alongside the normalized columns is deliberate: it gives auditors a verifiable original to compare against, satisfying the recordkeeping expectation in 21 CFR 1.1455 that records be maintained in their original captured form.

Minimal Reproducible Example

The failure is easiest to see with two realistic payloads: one with a naive timestamp and one with a whitespace-only lot code. Handing them directly to the driver reproduces both the hard exception and the silent-but-wrong persistence.

import psycopg2

conn = psycopg2.connect("dbname=trace user=ingest")
cur = conn.cursor()

# Payload A: naive timestamp into a TIMESTAMPTZ column.
# Postgres reinterprets it in the session timezone — event ordering silently shifts.
bad_ts = {
    "event_id": "CTE-001",
    "event_timestamp": "2024-03-15 08:30:00",  # no offset
    "lot_code": "LOT-88213",
}

# Payload B: whitespace-only lot code slips past NOT NULL.
empty_lot = {
    "event_id": "CTE-002",
    "event_timestamp": "2024-03-15T08:30:00Z",
    "lot_code": "   ",  # not NULL, so the constraint does not fire
}

for row in (bad_ts, empty_lot):
    cur.execute(
        "INSERT INTO fsma204_critical_tracking_events "
        "(event_id, kde_timestamp, lot_code) VALUES (%(event_id)s, "
        "%(event_timestamp)s, %(lot_code)s)",
        row,
    )
conn.commit()
# Result: payload A persists with a shifted timestamp; payload B persists an
# empty lot_code. Neither raises — the traceability chain is now corrupt.

Nothing in the driver path stops either record. The database enforces structure, not compliance semantics, so the fix has to move validation upstream of the insert.

The Fix: A Deterministic Validation Boundary

Decouple ingestion from persistence. An application-layer contract normalizes each payload, coerces types explicitly, and rejects non-compliant records before they reach SQL. The implementation below uses pydantic v2 for schema validation, structured logging for the audit trail, a circuit breaker to survive a supplier emitting malformed data at scale, and idempotent inserts so re-delivery is a no-op.

import logging
import hashlib
import json
import time
from datetime import datetime, timezone
from typing import Optional

import psycopg2
from pydantic import BaseModel, ValidationError, field_validator

logger = logging.getLogger("fsma204.kde_ingestion")
logger.setLevel(logging.INFO)


class KDEPayload(BaseModel):
    event_id: str
    event_timestamp: str
    lot_code: str
    product_description: str
    location_id: str
    raw_kde_json: Optional[dict] = None

    @field_validator("event_timestamp")
    @classmethod
    def normalize_iso8601(cls, v: str) -> str:
        # Strip timezone ambiguity and force UTC — the single most common
        # coercion defect. A naive string is rejected, never reinterpreted.
        try:
            dt = datetime.fromisoformat(v.replace("Z", "+00:00"))
        except ValueError as exc:
            raise ValueError(f"Invalid ISO 8601 timestamp: {v!r}") from exc
        if dt.tzinfo is None:
            raise ValueError(f"Timestamp missing timezone offset: {v!r}")
        return dt.astimezone(timezone.utc).isoformat()

    @field_validator("lot_code")
    @classmethod
    def enforce_lot_nonempty(cls, v: str) -> str:
        # Whitespace-only values pass NOT NULL but break lot lineage.
        if not v.strip():
            raise ValueError("lot_code cannot be empty or whitespace-only")
        return v.strip()


class CircuitBreaker:
    """Threshold-based breaker so one bad supplier cannot cascade."""

    def __init__(self, failure_threshold: int = 10, reset_timeout: int = 300):
        self.failure_count = 0
        self.threshold = failure_threshold
        self.reset_timeout = reset_timeout
        self.last_failure_time: Optional[float] = None

    def is_open(self) -> bool:
        if self.failure_count >= self.threshold:
            if self.last_failure_time and \
                    time.time() - self.last_failure_time < self.reset_timeout:
                return True
            self.failure_count = 0  # auto-reset after cooldown
        return False

    def record_success(self) -> None:
        self.failure_count = 0

    def record_failure(self) -> None:
        self.failure_count += 1
        self.last_failure_time = time.time()


def ingest_kde_record(conn, payload: dict, breaker: CircuitBreaker) -> dict:
    """Validate, normalize, and route a KDE payload to storage or quarantine."""
    if breaker.is_open():
        logger.warning("Circuit breaker open; routing to quarantine.")
        return {"status": "QUARANTINED", "reason": "CIRCUIT_BREAKER_OPEN"}

    cursor = None
    try:
        validated = KDEPayload(**payload)  # strict typing at the edge

        # Deterministic hash → deduplication key and audit anchor.
        payload_hash = hashlib.sha256(
            json.dumps(validated.model_dump(), sort_keys=True).encode()
        ).hexdigest()

        cursor = conn.cursor()
        cursor.execute(
            """
            INSERT INTO fsma204_critical_tracking_events
            (event_id, kde_timestamp, lot_code, product_description,
             location_id, raw_kde_json, payload_hash)
            VALUES (%s, %s, %s, %s, %s, %s, %s)
            ON CONFLICT (payload_hash) DO NOTHING
            """,
            (
                validated.event_id,
                validated.event_timestamp,
                validated.lot_code,
                validated.product_description,
                validated.location_id,
                json.dumps(validated.raw_kde_json or {}),
                payload_hash,
            ),
        )
        conn.commit()
        breaker.record_success()
        logger.info("Ingested KDE record %s", validated.event_id)
        return {"status": "ACCEPTED", "hash": payload_hash}

    except ValidationError as exc:
        conn.rollback()
        breaker.record_failure()
        failed = [err["loc"][0] for err in exc.errors()]
        logger.error("Schema validation failed; fields=%s", failed)
        return {"status": "REJECTED", "reason": "VALIDATION_ERROR",
                "failed_fields": failed}

    except psycopg2.Error as exc:
        conn.rollback()
        breaker.record_failure()
        logger.error("DB constraint violation %s: %s",
                     exc.diag.sqlstate, exc.diag.message_detail)
        return {"status": "REJECTED", "reason": "DB_CONSTRAINT",
                "sqlstate": exc.diag.sqlstate}

    finally:
        if cursor is not None:
            cursor.close()

Four decisions in this handler are compliance-relevant. The normalize_iso8601 validator rejects any timestamp without an explicit offset rather than guessing, which eliminates InvalidDatetimeFormat and preserves correct CTE ordering. Rejected payloads never touch the primary table — they return structured metadata (failed_fields, sqlstate, payload_hash) that a caller writes to a kde_quarantine table for reconciliation. The circuit breaker keeps a single misbehaving endpoint from exhausting worker capacity while the 24-hour export clock runs. And ON CONFLICT (payload_hash) DO NOTHING makes re-delivery of an identical payload idempotent, giving effectively-once persistence that keeps the append-only chain clean.

Verifying the Fix

Confirm the boundary behaves before trusting it in production. Three checks cover the failure modes above.

First, a unit assertion that a naive timestamp is rejected rather than coerced:

import pytest
from pydantic import ValidationError

def test_naive_timestamp_rejected():
    with pytest.raises(ValidationError):
        KDEPayload(
            event_id="CTE-001",
            event_timestamp="2024-03-15 08:30:00",  # no offset
            lot_code="LOT-88213",
            product_description="Romaine, chopped",
            location_id="0614141000005",
        )

def test_whitespace_lot_rejected():
    with pytest.raises(ValidationError):
        KDEPayload(
            event_id="CTE-002",
            event_timestamp="2024-03-15T08:30:00Z",
            lot_code="   ",
            product_description="Romaine, chopped",
            location_id="0614141000005",
        )

Second, confirm the ingestion path returns the right verdicts. A well-formed payload yields {"status": "ACCEPTED"}, and re-submitting the same payload returns ACCEPTED again while inserting no second row — the ON CONFLICT clause absorbs it.

Third, validate the persisted schema state directly. Every stored timestamp must carry UTC, and no lot code may be blank:

-- Should return zero rows if the boundary is enforcing the contract.
SELECT event_id, kde_timestamp, lot_code
FROM   fsma204_critical_tracking_events
WHERE  lot_code IS NULL
   OR  btrim(lot_code) = ''
   OR  kde_timestamp <> kde_timestamp AT TIME ZONE 'UTC' AT TIME ZONE 'UTC';

If that query returns rows, a write path is still bypassing the validator — usually a legacy batch job inserting directly rather than routing through ingest_kde_record.

Related Edge Cases to Check Next

• Transient transport failures masquerading as validation errors. A gateway timeout or HTTP 429 from a supplier is not a schema defect and should be retried with backoff, not quarantined. Separate transport retries from validation rejection using the discipline in API Polling Strategies before the payload reaches this boundary.
• Unauthenticated or spoofed submissions. A syntactically valid payload from an untrusted caller still corrupts provenance. Authenticate and tenant-scope the write at the network edge per Security Boundaries for Trace Data so only verified partners can insert CTEs.
• Divergent partner field shapes. When a distributor’s flat file maps to a different KDE shape than your internal ERP, the two feeds diverge at the ledger. Normalize both against one contract inside your Supplier Data Ingestion pipeline before they reach the SQL mapping stage.

Frequently Asked Questions

Related

• KDE Field Mapping Guide — the canonical field-level contract this SQL mapping enforces.
• Fallback Routing Logic — where quarantined and unrecoverable records go after rejection.
• Security Boundaries for Trace Data — authenticating and tenant-scoping writes before validation runs.
• API Polling Strategies — separating transient transport retries from schema rejection.
• Supplier Data Ingestion — normalizing divergent partner feeds onto one contract upstream.

Up: KDE Field Mapping Guide