Handling EDI parsing errors in pharmacy systems

A single malformed segment should never decide the fate of an entire EDI interchange. Yet that is exactly what most pharmacy ingestion code does: a naive parser wraps the whole file in one try/except, so one unparseable LIN segment aborts the 4,000 good ones behind it and the distributor feed stalls. The equal-and-opposite defect is worse — code that swallows the exception and keeps going, silently dropping a Schedule II quantity that an inspector will later find missing. Neither is acceptable when the records are governed by 21 CFR § 1304.21. This page solves the narrow, recurring problem of handling an EDI parse error correctly: classifying it, isolating it without collateral damage, and routing it to retry or to a compliance hold based on whether the failure is transient or permanent.

It is a focused recipe within the Error Handling & Retry Mechanisms cluster, which sits under the broader Data Ingestion & Inventory Sync Workflows architecture. The parser that produces these errors — the loop-aware reader for X12 852 and 846 transactions — is documented in the EDI 852 & 846 Parsing Pipelines cluster; this page picks up the moment that parser raises.

Failure Modes: Classifying an EDI Parse Error

Pharmacy EDI failures do not arrive as a single exception type. They cluster into three disposition classes, and the disposition — not the symptom — is what your handler must key on. A transient failure is safe to retry; a permanent semantic failure must never be retried because it will fail identically forever; a compliance failure must halt and escalate rather than retry at all.

The single most important rule: a RETRY and a DISCARD must be distinguishable programmatically, because retrying a semantically invalid NDC is an infinite loop, and discarding a transient truncation is silent data loss. The classifier below makes that decision explicit and records the rationale with a tamper-evident hash.

Prerequisites & Environment

This recipe targets Python 3.11+. It depends only on pydantic (v2) plus the standard library:

You should already understand the X12 852/846 envelope (ISA/GS/ST … SE/GE/IEA) and the item loop (LIN/QTY/DTM). Two pharmacy-specific rules are consumed, not re-derived here: the canonical NDC conversion in NDC-11 vs NDC-10 Parsing Standards, and the scheduling lookup performed by the DEA Schedule II-V Classification Mapping engine, which is what tells the classifier a LIN is Schedule II and therefore subject to the COMPLIANCE_HOLD path.

Implementation: Per-Segment Classifier with Quarantine Routing

The handler is a pure function over one segment. It never raises on bad input — instead it returns either a parsed dict or a frozen QuarantineRecord whose disposition field tells the caller exactly what to do next. Every quarantine record carries the SHA-256 of the exact bytes that failed, so the rejection is itself an auditable event under 21 CFR Part 11.

from __future__ import annotations

import hashlib
import logging
import re
from datetime import datetime, timezone
from enum import Enum
from typing import Optional

from pydantic import BaseModel, ConfigDict, Field

logging.basicConfig(level=logging.INFO, format="%(message)s")
logger = logging.getLogger("edi.error_handler")

# Anchored on both ends so a malformed line cannot trigger catastrophic
# backtracking. FDA 5-4-2 canonical NDC form.
NDC_PATTERN = re.compile(r"^\d{5}-\d{4}-\d{2}$")
ALLOWED_UOM = frozenset({"EA", "BX", "PK", "CA", "ML", "GM"})


class ErrorTaxonomy(str, Enum):
    TRANSPORT = "transport_truncation"
    SYNTAX = "syntax_error"
    NDC_INVALID = "ndc_invalid"
    UOM_MISMATCH = "uom_mismatch"
    SCHEDULE_LOT_MISSING = "schedule_lot_missing"
    DUPLICATE_CONTROL = "duplicate_control_number"


class Disposition(str, Enum):
    RETRY = "retry"               # transient: re-fetch / backoff
    DISCARD = "discard"           # permanent semantic: quarantine, never retry
    COMPLIANCE_HOLD = "hold"      # halt, escalate to pharmacist review


# A parse error's disposition is a property of its taxonomy, not its symptom.
DISPOSITION_MAP: dict[ErrorTaxonomy, Disposition] = {
    ErrorTaxonomy.TRANSPORT: Disposition.RETRY,
    ErrorTaxonomy.SYNTAX: Disposition.RETRY,
    ErrorTaxonomy.NDC_INVALID: Disposition.DISCARD,
    ErrorTaxonomy.UOM_MISMATCH: Disposition.DISCARD,
    ErrorTaxonomy.SCHEDULE_LOT_MISSING: Disposition.COMPLIANCE_HOLD,
    ErrorTaxonomy.DUPLICATE_CONTROL: Disposition.COMPLIANCE_HOLD,
}


class QuarantineRecord(BaseModel):
    """Frozen, PHI-free audit record for a rejected segment."""
    model_config = ConfigDict(frozen=True)

    transaction_set_id: str
    segment_sequence: int
    error_taxonomy: ErrorTaxonomy
    disposition: Disposition
    raw_segment: str
    payload_hash: str
    timestamp_utc: str = Field(
        default_factory=lambda: datetime.now(timezone.utc).isoformat(timespec="milliseconds")
    )


def _sha256(payload: str) -> str:
    return hashlib.sha256(payload.encode("utf-8")).hexdigest()


def _quarantine(tx_id: str, seq: int, taxonomy: ErrorTaxonomy, raw: str) -> QuarantineRecord:
    record = QuarantineRecord(
        transaction_set_id=tx_id,
        segment_sequence=seq,
        error_taxonomy=taxonomy,
        disposition=DISPOSITION_MAP[taxonomy],
        raw_segment=raw,
        payload_hash=_sha256(raw),
    )
    # Structured, PHI-free audit line. The raw segment carries no patient data.
    logger.info(record.model_dump_json())
    return record


def classify_segment(
    segment: str,
    tx_id: str,
    sequence: int,
    is_scheduled: bool = False,
) -> tuple[Optional[dict], Optional[QuarantineRecord]]:
    """Parse one X12 segment, or quarantine it with an explicit disposition.

    `is_scheduled` is the result of the DEA schedule lookup for this LIN; when
    True, a missing lot/expiry escalates to COMPLIANCE_HOLD instead of DISCARD.
    """
    parts = segment.strip().split("*")
    if len(parts) < 2:
        # Truncated / structurally empty: treat as transient transport loss.
        return None, _quarantine(tx_id, sequence, ErrorTaxonomy.TRANSPORT, segment)

    seg_id = parts[0]

    if seg_id == "LIN":
        ndc = parts[3] if len(parts) > 3 else ""
        if not NDC_PATTERN.match(ndc):
            return None, _quarantine(tx_id, sequence, ErrorTaxonomy.NDC_INVALID, segment)
        # Schedule II-V items must carry lot + expiry to satisfy 21 CFR 1304.21.
        if is_scheduled and len(parts) < 6:
            return None, _quarantine(
                tx_id, sequence, ErrorTaxonomy.SCHEDULE_LOT_MISSING, segment
            )

    if seg_id == "QTY":
        uom = parts[3] if len(parts) > 3 else ""
        if uom not in ALLOWED_UOM:
            return None, _quarantine(tx_id, sequence, ErrorTaxonomy.UOM_MISMATCH, segment)

    return {"segment_id": seg_id, "transaction_set_id": tx_id, "sequence": sequence}, None

The caller drives disposition off the returned record rather than inspecting the exception text:

def handle(segment: str, tx_id: str, seq: int, is_scheduled: bool) -> str:
    parsed, rejected = classify_segment(segment, tx_id, seq, is_scheduled)
    if rejected is None:
        return "committed"
    if rejected.disposition is Disposition.RETRY:
        return "re_enqueued"          # hand to exponential-backoff retry
    if rejected.disposition is Disposition.COMPLIANCE_HOLD:
        return "escalated"            # block the interchange, page a pharmacist
    return "quarantined"              # DISCARD: dead-letter, never retry

A RETRY disposition should feed the backoff machinery and dead-letter queue defined in the parent Error Handling & Retry Mechanisms cluster; under a disconnected POS terminal, the same retry is deferred through Fallback Routing for Offline Sync so a transient failure is not mistaken for a permanent one.

Verification & Testing

Correctness is binary for each disposition: a transient truncation must be RETRY, an invalid NDC must be DISCARD, and a scheduled item missing its lot must be COMPLIANCE_HOLD. A small table-driven test pins all three so a future refactor of DISPOSITION_MAP cannot silently re-route a controlled-substance failure into the retry loop.

def test_dispositions():
    # Transient: structurally empty segment -> retry.
    _, rej = classify_segment("ZZ", "0001", 1)
    assert rej.disposition is Disposition.RETRY

    # Semantic: malformed NDC -> discard, never retry.
    _, rej = classify_segment("LIN**N4*123*XX", "0001", 2)
    assert rej.error_taxonomy is ErrorTaxonomy.NDC_INVALID
    assert rej.disposition is Disposition.DISCARD

    # Compliance: Schedule II LIN with no lot/expiry -> hold.
    _, rej = classify_segment("LIN**N4*00093-7214-10", "0001", 3, is_scheduled=True)
    assert rej.disposition is Disposition.COMPLIANCE_HOLD

    # Happy path: valid NDC, scheduled, lot + expiry present -> parsed.
    parsed, rej = classify_segment(
        "LIN**N4*00093-7214-10*LOT*A1234", "0001", 4, is_scheduled=True
    )
    assert rej is None and parsed["segment_id"] == "LIN"

For the compliance dimension, inspect the structured log. Each rejection writes one JSON line whose payload_hash is the SHA-256 of the exact bytes that failed — an inspector can recompute it from the retained raw_segment and confirm the rejection record was never altered:

{"transaction_set_id":"0001","segment_sequence":3,"error_taxonomy":"schedule_lot_missing","disposition":"hold","raw_segment":"LIN**N4*00093-7214-10","payload_hash":"a3f1...","timestamp_utc":"2026-06-28T14:02:11.004+00:00"}

Records destined for the controlled-substance ledger should additionally pass a structural gate before they are trusted — see JSON Schema Validation for Drug Records — so that a segment which parsed but is semantically thin is caught before commitment, not after.

Gotchas & Compliance Pitfalls

• Retrying a semantic failure is an infinite loop. An invalid NDC will fail identically on every attempt. The disposition split exists precisely so DISCARD errors go straight to the dead-letter queue; never wire them into backoff. Mixing the two is the most common way a “self-healing” pipeline burns retry budget and masks real data loss.
• A COMPLIANCE_HOLD must block, not skip. When a Schedule II LIN lacks lot/expiry, the correct action is to halt the interchange and escalate to pharmacist review — not to quarantine the one segment and commit the rest. Committing the surrounding transaction set leaves a controlled-substance record incomplete, which is the exact drift 21 CFR § 1304.21 is meant to prevent.
• Ambiguous NDC padding masquerades as a parse error. A code that fails the strict 5-4-2 matcher may be a valid 4-4-2 or 5-3-2 form, not garbage. Resolve it through the directional rules in NDC-11 vs NDC-10 Parsing Standards before classifying it as NDC_INVALID; discarding a recoverable code is silent loss.
• Never mutate raw EDI in place. Recovery generates a corrected derivative linked to the original payload_hash; the original bytes stay immutable. An audit chain that rewrites its own source cannot prove integrity.
• Default UOM is a trap. Falling back to EA for an unknown unit produces perpetual reconciliation drift. Fail the segment to DISCARD instead and let a human map the unit — a wrong base-unit count on a controlled substance is far costlier than a quarantined line.
• Quarantine is only real if something reads it. A DISCARD queue that no one drains is an unreported gap. Wire it to the deferred-review path and dispatch recovered records through Async Batch Processing for Inventory Updates so a replayed segment cannot double-post.

Frequently Asked Questions

Related

• Error Handling & Retry Mechanisms — parent cluster covering backoff, idempotency, and the dead-letter queue these dispositions feed.
• EDI 852 & 846 Parsing Pipelines — the parser whose failures this handler classifies.
• Parsing EDI 852 files with Python pandas — the loop-aware reader that produces the quarantine queue.
• NDC-11 vs NDC-10 Parsing Standards — resolve recoverable NDC formats before classifying them as invalid.
• Fallback Routing for Offline Sync — deferred retry path when the POS terminal is disconnected.