CSV/EDI Parser Setup for FSMA 204 KDE Ingestion

FSMA 204 compliance is not achieved by collecting data; it is achieved by capturing Key Data Elements (KDEs) deterministically at every Critical Tracking Event (CTE). The specific engineering problem this page solves is the one that breaks most supplier feeds before validation ever runs: heterogeneous file formats. Modern supplier ecosystems rarely standardize on a single interchange format. A single distributor may send flat CSV exports on Monday, EDI 856 Advance Ship Notices (ASNs) from a Tier-1 manufacturer on Tuesday, and proprietary delimited dumps from a legacy ERP on Wednesday — each with different encodings, header names, and date conventions. The parser is the enforcement point where all of that fragmentation must collapse into one canonical KDE record, or the lot chain fractures before it ever reaches the ledger.

Within the broader Supplier Data Ingestion & Sync Automation pipeline, this parser is the ingestion vector for every supplier that does not expose a modern REST endpoint. The triggering requirement is concrete and mandated: a shipping CTE captured under 21 CFR 1.1340 is only defensible if the parser extracts its traceability lot code, quantity, unit of measure, ship-from and receive-to GLNs, and event timestamp with zero ambiguity. A parser that silently drops a malformed field or guesses at a date format introduces compliance risk that compounds during recall scoping. This page details the routing architecture, the KDE data contract, a runnable pydantic v2 implementation, the quarantine strategy that keeps rejected records audit-visible, and the operational configuration needed to run it in production.

Parser Architecture & Routing Strategy

The ingestion layer must operate as a stateless transformation engine. Raw payloads land in an immutable staging bucket, where a routing dispatcher identifies the format via MIME type, file extension, or magic bytes. Once classified, the payload is routed to a format-specific extractor. Each extractor applies the same strict KDE contract, validates structural integrity, and emits normalized records. When suppliers instead deliver files over SFTP or expose pollable endpoints, this parser pairs with API Polling Strategies to guarantee idempotent fetch cycles, enforce exponential backoff, and prevent duplicate processing — the same KDE contract, a different transport.

Figure — Parser ingestion pipeline:

The non-negotiable requirement is deterministic mapping. Every traceability lot code (TLC), product description, quantity, unit of measure, ship-from/receive-to GLN, and timestamp must survive transformation with a single, unambiguous target field. FSMA 204 mandates that KDEs remain intact across handoffs; the routing dispatcher’s job is to guarantee that a CSV row and an EDI 856 HL loop describing the same shipment produce byte-for-byte identical canonical records. Any format-specific quirk — a BOM-prefixed CSV header, an X12 composite element, a European DD.MM.YYYY date — must be resolved inside the extractor and never leak downstream.

The KDE Data Contract & Field Mapping

Before implementing extraction logic, define the immutable KDE contract every extractor must satisfy. The table below maps the raw supplier keys the parser encounters onto their canonical KDE fields, states the validation rule applied at the ingestion boundary, and cites the specific mandate in 21 CFR Part 1, Subpart S. This is the parser-scoped view of the complete field catalog documented in the KDE Field Mapping Guide, which covers every transport.

Per the GS1 Global Location Number (GLN) Standard, GLNs must be exactly 13 numeric digits and pass modulo-10 check-digit validation. Validation must occur at the schema level, not post-hoc: use strict type coercion, reject ambiguous date formats, and reject any GLN that fails the check digit. Optional KDEs should default to an explicit null rather than an empty string, so downstream queries can distinguish “not applicable” from “missing and required.” Malformed records are never silently coerced — they are quarantined with full context for manual reconciliation.

Production Implementation

The following implementation demonstrates a production-ready parser built on a pydantic v2 KDE model. It leverages field_validator for boundary enforcement, structured audit logging, tenacity for resilient retry orchestration, and explicit quarantine routing. Records that fail validation are isolated with full provenance rather than dropped. The code assumes payloads arrive as raw bytes and focuses on deterministic extraction and audit-trail generation.

from __future__ import annotations

import csv
import io
import json
import logging
import hashlib
from decimal import Decimal, InvalidOperation
from datetime import datetime, timezone
from enum import Enum
from pathlib import Path
from typing import Any

from pydantic import BaseModel, field_validator, ValidationError
from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential,
    retry_if_exception_type,
)

# Audit-ready structured logging configuration
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
    handlers=[logging.StreamHandler()],
)
logger = logging.getLogger("fsma204_parser")


class UnitOfMeasure(str, Enum):
    CASE = "case"
    LB = "lb"
    KG = "kg"
    EA = "ea"
    PALLET = "pallet"
    LITER = "liter"
    GALLON = "gallon"


class CTEType(str, Enum):
    HARVESTING = "harvesting"
    COOLING = "cooling"
    INITIAL_PACKING = "initial_packing"
    SHIPPING = "shipping"
    RECEIVING = "receiving"
    TRANSFORMATION = "transformation"


def gln_check_digit_valid(gln: str) -> bool:
    """Validate a 13-digit GS1 GLN via its mod-10 check digit."""
    if len(gln) != 13 or not gln.isdigit():
        return False
    digits = [int(c) for c in gln]
    # Weights alternate 1,3,... from the left across the first 12 digits.
    total = sum(d * (3 if i % 2 else 1) for i, d in enumerate(digits[:12]))
    check = (10 - (total % 10)) % 10
    return check == digits[12]


class KDERecord(BaseModel):
    """Immutable FSMA 204 Key Data Element record enforced at the boundary."""

    model_config = {"frozen": True}

    traceability_lot_code: str
    product_description: str
    quantity: Decimal
    unit_of_measure: UnitOfMeasure
    ship_from_gln: str
    receive_to_gln: str
    event_timestamp: datetime
    event_type: CTEType
    payload_hash: str

    @field_validator("traceability_lot_code", "product_description")
    @classmethod
    def non_empty(cls, v: str) -> str:
        v = v.strip()
        if not v:
            raise ValueError("required KDE is empty")
        if len(v) > 64:
            raise ValueError("value exceeds 64 characters")
        return v

    @field_validator("quantity")
    @classmethod
    def positive_quantity(cls, v: Decimal) -> Decimal:
        if v <= 0:
            raise ValueError(f"quantity must be > 0, got {v}")
        return v

    @field_validator("ship_from_gln", "receive_to_gln")
    @classmethod
    def valid_gln(cls, v: str) -> str:
        v = v.strip()
        if not gln_check_digit_valid(v):
            raise ValueError(f"invalid GLN (13 digits + mod-10 check): '{v}'")
        return v

    @field_validator("event_timestamp")
    @classmethod
    def utc_not_future(cls, v: datetime) -> datetime:
        if v.tzinfo is None:
            raise ValueError("event_timestamp must be timezone-aware")
        v = v.astimezone(timezone.utc)
        if v > datetime.now(timezone.utc):
            raise ValueError(f"event_timestamp is in the future: {v.isoformat()}")
        return v


class FSMA204Parser:
    def __init__(self, quarantine_dir: Path = Path("./quarantine")) -> None:
        self.quarantine_dir = quarantine_dir
        self.quarantine_dir.mkdir(exist_ok=True)

    def _payload_hash(self, raw_bytes: bytes) -> str:
        return hashlib.sha256(raw_bytes).hexdigest()

    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=2, max=10),
        retry=retry_if_exception_type((csv.Error, UnicodeDecodeError)),
    )
    def parse_csv(
        self, raw_bytes: bytes, column_mapping: dict[str, str]
    ) -> list[KDERecord]:
        """Extract, map, and validate CSV rows into canonical KDE records.

        `column_mapping` maps source CSV headers to canonical KDE field names,
        e.g. {"LotID": "traceability_lot_code", "ShipGLN": "ship_from_gln"}.
        Transport faults (encoding, dialect) are retried; row-level validation
        failures are quarantined, never retried.
        """
        payload_hash = self._payload_hash(raw_bytes)
        # utf-8-sig transparently strips a byte-order mark if present.
        text = io.TextIOWrapper(io.BytesIO(raw_bytes), encoding="utf-8-sig")
        reader = csv.DictReader(text)

        if not reader.fieldnames:
            raise csv.Error("empty or malformed CSV header detected")

        records: list[KDERecord] = []
        for row_num, row in enumerate(reader, start=2):  # header is row 1
            mapped = {
                target: (row.get(source) or "").strip()
                for source, target in column_mapping.items()
            }
            try:
                mapped["payload_hash"] = payload_hash
                records.append(KDERecord.model_validate(mapped))
            except (ValidationError, ValueError, InvalidOperation) as exc:
                logger.warning("Row %d failed KDE validation: %s", row_num, exc)
                self._quarantine(row_num, row, str(exc), payload_hash)

        logger.info(
            "Parsed %d/%d valid KDE records from payload %s",
            len(records), row_num - 1, payload_hash[:12],
        )
        return records

    def parse_edi_856(self, raw_bytes: bytes) -> list[KDERecord]:
        """Route an EDI 856 ASN to the dedicated X12 segment extractor.

        Full extraction requires ISA/GS interchange traversal and HL-loop
        walking; see the EDI 856 guide linked below for the state machine.
        """
        payload_hash = self._payload_hash(raw_bytes)
        logger.info("Routing EDI 856 payload %s to X12 extractor", payload_hash[:12])
        # In production, delegate to a dedicated X12 segment parser that emits
        # the same KDERecord objects, keeping the contract transport-agnostic.
        return []

    def _quarantine(
        self, row_num: int, row: dict[str, Any], error: str, payload_hash: str
    ) -> None:
        record = {
            "row_index": row_num,
            "raw_data": row,
            "validation_error": error,
            "payload_fingerprint": payload_hash,
            "quarantined_at_utc": datetime.now(timezone.utc).isoformat(),
        }
        q_file = self.quarantine_dir / f"quarantine_{payload_hash[:8]}_row{row_num}.json"
        q_file.write_text(json.dumps(record, indent=2))
        logger.error("Quarantined row %d to %s", row_num, q_file)

Because KDERecord is the single validation gate, a CSV row and a future EDI 856 HL loop both converge on the same model — the contract is transport-agnostic by construction. The payload_hash embedded in every record and every quarantine artifact is the cryptographic fingerprint that ties a persisted KDE back to the exact byte stream a supplier delivered.

Error Handling & Quarantine Strategy

The parser draws a hard line between two failure classes, and handles each differently. Transport faults — a corrupt byte-order mark, an unexpected CSV dialect, a truncated read — are transient and non-deterministic, so they are wrapped by tenacity and retried with exponential backoff up to three attempts. Validation faults — a GLN that fails its check digit, a negative quantity, a timezone-naive timestamp — are deterministic: retrying them changes nothing, so they are routed straight to quarantine with zero retries. Conflating the two either wastes cycles retrying a permanently malformed row or, worse, silently discards a transient failure that a retry would have recovered.

Quarantine is not merely error handling; it is a regulatory artifact. Each quarantined row is written as a self-contained JSON document carrying the original raw fields, the precise validation error, the payload 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 parser processes rows independently, so one malformed row never aborts the batch — the valid records in the same file continue to the ledger while the bad row is isolated for reconciliation. This partial-commit contract mirrors the dead-letter behavior of the shared Error Handling Workflows, so operators reconcile parser rejections through the same tooling as every other ingestion stage.

Quarantine depth per supplier is itself a signal. A sudden spike in rejected rows from one vendor usually means an upstream schema change — a renamed header, a switched date format, a new GLN prefix — and that anomaly is surfaced to the Data Quality Monitoring layer against per-supplier SLAs rather than left to accumulate silently.

Integration with the Ingestion Pipeline

This parser is one ingestion vector inside the parent Supplier Data Ingestion & Sync Automation pipeline, and it is deliberately narrow: its only job is to detect, extract, validate, and hand off. Validated KDERecord objects flow into the message queue that fronts Async Batch Processing, where I/O-bound persistence and enrichment run on a worker pool decoupled from parsing. For high-volume supplier feeds, synchronous parse-and-write quickly becomes a bottleneck; offloading heavy validation and database writes to workers maintains backpressure and prevents memory exhaustion when a single file carries hundreds of thousands of rows.

The contract this parser enforces is identical to the one applied by the Schema Validation Rules at every other ingestion boundary, so a flat-file record and a REST-polled record are indistinguishable once they reach the ledger. Because the parser reads immutable staging bytes and emits fingerprinted records, its output also 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.

When the payload is X12 rather than delimited text, the extraction logic diverges sharply from CSV row iteration. EDI 856 files require traversing the ISA/GS interchange and functional-group envelopes, walking the hierarchical HL loops with their BSN, PRF, and LIN segments, handling segment terminators, and mapping composite elements to KDEs. For the full state-machine implementation of ISA/GS headers, BSN segments, and HL-loop traversal, follow the dedicated guide on parsing EDI 856 for FSMA compliance. The Python csv module handles delimited text efficiently, but EDI requires a state machine or a dedicated X12 library to maintain segment context across line breaks.

Operational Notes

Deploy the parser as a triggered job — an object-storage event handler, a Kubernetes Job, or a Celery task fired on file arrival — rather than a long-lived loop, so a crashed run is restarted cleanly by the scheduler and immutable staging 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_validate 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 STAGING_BUCKET for immutable raw payloads, a QUARANTINE_STORE target (the local ./quarantine directory in the example exists only to make the module runnable — use versioned, access-controlled object storage in production), and the per-supplier column_mapping that translates each vendor’s headers into canonical KDE names. Store quarantine artifacts with a minimum two-year retention policy to satisfy 21 CFR record-retention expectations, and align log retention and accessibility with the FDA FSMA 204 Traceability Rule. Keep each supplier’s column_mapping under version control so a schema change is a reviewable diff, not a silent production surprise.

Frequently Asked Questions

Related

• Parsing EDI 856 for FSMA compliance — the ISA/GS interchange and HL-loop state machine that emits the same KDE contract.
• API Polling Strategies — the REST ingestion vector for suppliers without flat-file exports.
• Async Batch Processing — the worker pool that persists validated parser output at volume.
• Schema Validation Rules — the shared KDE enforcement applied at every ingestion boundary.
• Error Handling Workflows — dead-letter routing and operator reconciliation for quarantined rows.

Up: Supplier Data Ingestion & Sync Automation — this parser is the flat-file and EDI ingestion vector of the parent pipeline.