Automating DSCSA Compliance Gap Checks with Python

A compliance gap is any serialized event that reaches your repository missing a required data element, arriving out of chronological order, or resolving to a failed verification — and under the Drug Supply Chain Security Act (DSCSA), every one of those gaps is a latent suspect-product hold waiting to fire. This page is a concrete, code-first solution to one narrow problem within Suspect Product Investigation Workflows: how to scan an inbound stream of GS1 EPCIS events in Python and deterministically flag the gaps — malformed GTINs, timestamp drift between trading partners, and Verification Router Service mismatches — before they propagate into the commercial stream, inflate false-positive quarantines, or surface as a data-integrity finding in an FDA inspection. The goal is a validation stage that either passes an event cleanly or emits a signed, structured gap record that an investigation workflow can act on without any manual triage.

Figure — The gap-check pipeline: one inbound EPCIS event fans out to three deterministic validators, merges, and either commits clean or emits a signed gap record.

Prerequisites #

• Python 3.10+ — the snippets use list[dict] generics, X | Y union syntax, and datetime.timezone.utc.
• Pydantic v2 — strict schema validation with field_validator; this is the same validator style used across the site’s Schema Validation & Error Handling patterns.
• polars — vectorized timestamp comparison across millions of events without a Python-level loop.
• aiohttp + tenacity — concurrent, retry-guarded calls to your Verification Router Service Architecture endpoints.
• cryptography — X.509 signature verification for EPCIS documents, aligned with your Data Security & Encryption Boundaries controls.
• DSCSA data prerequisites — read access to the L4 EPCIS 2.0 repository, an active SGTIN pool (GTIN (01) + serial (21)), trading-partner GLNs, and each partner’s agreed clock-skew tolerance (commonly ±15 minutes).

Before writing a validator, name the gap taxonomy you are checking against, because each class maps to a distinct DSCSA obligation: data-completeness and format violations (missing GTIN, serial, lot (10), or expiry (17); a GTIN failing GS1 modulo-10); event-sequencing and temporal drift (a shipping event timestamped before commissioning); VRS routing and status failures (a serial routed to a decommissioned endpoint, or a false verification with reason code no_match, expired, or recalled); aggregation and pedigree breaks (a child SGTIN with no parent, handled in depth under Parent-Child Serial Mapping); and cryptographic-boundary failures (an unsigned or tampered EPCIS document). The steps below cover the first three — the ones a Python stage can resolve at ingestion time — and close the loop with signature and audit checks.

Step-by-Step Solution #

Step 1 — Enforce GTIN check-digit and serial format at the schema edge #

Satisfies: GS1 General Specifications identifier structure — a 14-digit GTIN whose final digit is the modulo-10 check digit, and a serial reference of 1–20 characters per the SGTIN construction rules in GS1 Standards Implementation. Reject non-conforming identifiers before they ever reach the repository.

from pydantic import BaseModel, field_validator
import re

class SerializedIdentifier(BaseModel):
    gtin: str
    serial: str

    @field_validator("gtin")
    @classmethod
    def validate_gtin_check_digit(cls, v: str) -> str:
        if not re.match(r"^\d{14}$", v):
            raise ValueError("GTIN must be exactly 14 numeric digits.")
        # GS1 modulo-10: weight the 13 data digits right-to-left, alternating 3/1
        digits = [int(d) for d in reversed(v[:-1])]
        check = (10 - sum(d * (3 if i % 2 == 0 else 1) for i, d in enumerate(digits))) % 10
        if check != int(v[-1]):
            raise ValueError("GTIN check-digit validation failed.")
        return v

    @field_validator("serial")
    @classmethod
    def validate_serial(cls, v: str) -> str:
        if not re.match(r"^[A-Za-z0-9]{1,20}$", v):
            raise ValueError("Serial must be alphanumeric, max 20 chars.")
        return v

A raised ValidationError here is not an exception to swallow — it is a FORMAT_VIOLATION gap. Catch it at the call site and convert it into a structured gap record (Step 5) rather than letting it crash the batch.

Step 2 — Detect temporal drift and out-of-order events #

Satisfies: EPCIS event-sequencing integrity — commissioning (ObjectEvent, bizStep commissioning) must precede aggregation, which must precede shipping. An eventTime that moves backward for the same serial, or jumps forward past the trading-partner skew tolerance, is a TEMPORAL_DRIFT gap. polars runs this comparison vectorized across the whole batch.

import polars as pl
from datetime import timedelta

def detect_temporal_drift(events_df: pl.DataFrame,
                          skew_minutes: int = 15) -> pl.DataFrame:
    """Flag consecutive events for one serial that go backward in time
    or jump forward past the trading-partner skew threshold."""
    events_df = events_df.with_columns(
        pl.col("eventTime").str.to_datetime(
            format="%Y-%m-%dT%H:%M:%S%.fZ", strict=False)
    )
    drift = (
        events_df
        .sort(["serial", "eventTime"])
        .with_columns(pl.col("eventTime").diff().over("serial").alias("delta"))
    )
    threshold = timedelta(minutes=skew_minutes)
    return drift.filter(
        (pl.col("delta") < pl.lit(timedelta(0)))
        | (pl.col("delta") > pl.lit(threshold))
    )

Always parse eventTime as timezone-aware UTC. EPCIS carries a separate eventTimeZoneOffset field; if you compare naive local timestamps across partners in different zones, you will manufacture drift that does not exist.

Step 3 — Resolve VRS status with bounded concurrency and retries #

Satisfies: the DSCSA verification obligation — a saleable-return or suspect unit must be verified against the manufacturer’s response service. Network flakiness must not be misread as a compliance gap, so wrap each lookup in exponential backoff and cap concurrency so a large batch cannot flood the endpoint.

import aiohttp, asyncio
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3),
       wait=wait_exponential(multiplier=1, min=2, max=10))
async def query_vrs(session: aiohttp.ClientSession,
                    gtin: str, serial: str) -> dict:
    url = "https://vrs-endpoint.example.com/verify"
    async with session.get(url, params={"gtin": gtin, "serial": serial}) as r:
        r.raise_for_status()
        return await r.json()

async def batch_verify(identifiers: list[dict]) -> list[dict]:
    sem = asyncio.Semaphore(10)  # cap concurrent VRS requests
    async with aiohttp.ClientSession() as session:
        async def bounded(item: dict) -> dict:
            async with sem:
                return await query_vrs(session, item["gtin"], item["serial"])
        results = await asyncio.gather(
            *[bounded(i) for i in identifiers], return_exceptions=True)
    # Surface only non-VALID responses for quarantine routing
    return [r for r in results if isinstance(r, dict) and r.get("status") != "VALID"]

Distinguish a transport failure (all retries exhausted → route to manual review, do not clear the unit) from a verification failure (status != VALID → a VRS_INVALID gap). Silently treating a timeout as “not invalid” is how illegitimate product slips through.

Step 4 — Verify the EPCIS document signature #

Satisfies: DSCSA data-integrity and the ALCOA+ “Original/Attributable” principles — an interoperable EPCIS document exchanged between partners should be digitally signed with an X.509 certificate over TLS 1.3. A failed signature is a SIGNATURE_INVALID gap that voids the pedigree regardless of payload contents.

from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.serialization import load_pem_public_key
from cryptography.hazmat.primitives.asymmetric import padding

def verify_epcis_signature(payload: bytes, signature: bytes,
                           public_key_pem: bytes) -> bool:
    public_key = load_pem_public_key(public_key_pem)
    try:
        public_key.verify(signature, payload,
                          padding.PKCS1v15(), hashes.SHA256())
        return True
    except Exception:
        return False

Step 5 — Emit a signed, structured gap record #

Satisfies: the six-year DSCSA retention and inspector-facing traceability requirement — every gap decision becomes an immutable, hash-anchored record that an investigation workflow can consume and an auditor can replay. Standardize the shape so downstream routing is deterministic.

import hashlib, json
from datetime import datetime, timezone

def build_gap_record(event: dict, gap_type: str, severity: str) -> dict:
    epc_hash = hashlib.sha256(
        json.dumps(event, sort_keys=True).encode()).hexdigest()
    action = {"CRITICAL": "QUARANTINE",
              "MEDIUM": "MANUAL_REVIEW"}.get(severity, "AUTO_RESOLVE")
    return {
        "event_id": event.get("eventID"),
        "epc_hash": epc_hash,
        "gap_type": gap_type,           # FORMAT_VIOLATION | TEMPORAL_DRIFT | VRS_INVALID ...
        "severity": severity,           # LOW | MEDIUM | CRITICAL
        "recommended_action": action,   # QUARANTINE | MANUAL_REVIEW | AUTO_RESOLVE
        "detected_at": datetime.now(timezone.utc).isoformat(),
    }

Publish these records to a durable channel — a Kafka topic or a SELECT ... FOR UPDATE-guarded ledger table — so a QUARANTINE action opens an investigation case with the evidence package already attached, and the packaging line never has to stop for triage.

Verification #

Confirm the pipeline behaves before you point it at production traffic. The fastest signal is a table-driven unit test that feeds one known-good and one known-bad case through each validator:

import pytest
from pydantic import ValidationError

def test_gtin_check_digit_rejects_bad_digit():
    with pytest.raises(ValidationError):
        SerializedIdentifier(gtin="00312345678900", serial="ABC123")  # wrong check digit

def test_gtin_check_digit_accepts_valid():
    ok = SerializedIdentifier(gtin="00312345678906", serial="ABC123")
    assert ok.gtin.endswith("6")

def test_temporal_drift_flags_backward_event():
    import polars as pl
    df = pl.DataFrame({
        "serial": ["S1", "S1"],
        "eventTime": ["2026-07-01T10:00:00.000Z", "2026-07-01T09:00:00.000Z"],
    })
    assert detect_temporal_drift(df).height == 1

Beyond unit tests, run three checks against real output: confirm your emitted EPCIS still validates against the GS1 EPCIS 2.0 Standard schema; recompute epc_hash from a stored gap record and confirm it matches the source payload byte-for-byte (proof the audit trail is tamper-evident); and reconcile the count of QUARANTINE records against opened investigation cases so no critical gap is silently dropped. Retention windows and record formats trace back to the FDA’s DSCSA guidance for industry.

Gotchas & Edge Cases #

• Leading zeros and NDC packaging. A GTIN embeds the NDC, and stripping a leading zero when mapping between NDC formats silently breaks the check digit. Treat GTINs as opaque 14-character strings — never as integers.
• UTC vs. local time in eventTime. EPCIS stores eventTime in UTC alongside a separate eventTimeZoneOffset. Comparing naive local timestamps across partners fabricates drift; always normalize to timezone-aware UTC first.
• VRS timeouts masquerading as clean. An exhausted-retry transport error is not a VALID response. Route it to manual review; do not let return_exceptions=True results fall through as “passed.”
• Non-deterministic hashing. Serialize payloads with sort_keys=True before hashing. Unordered JSON keys produce different epc_hash values for the same event and destroy replayability of the audit trail.
• Idempotency-key collisions on replay. Message brokers redeliver. Key gap records on (event_id, gap_type) so a redelivered event does not open a duplicate investigation case.

Related #

• Up to the parent section: Suspect Product Investigation Workflows
• Verification Router Service Architecture — the endpoints Step 3 queries
• Schema Validation & Error Handling — deeper Pydantic EPCIS validation patterns
• DSCSA Compliance Architecture & Standards Mapping — the standards this pipeline enforces