Schema Validation & Error Handling

Schema validation and error handling is the contract-enforcement stage of the ingestion pipeline: the layer that decides whether a payload arriving from an analyzer is structurally trustworthy enough to become a clinical record, or whether it must be quarantined for human adjudication. Its remit is narrow and unforgiving. It does not decide whether a potassium of 6.9 mmol/L is critical — that is a downstream clinical decision — but it does decide whether the value is a parseable number in the right OBX field, carried by a well-formed message with the mandatory identifiers intact. Everything past this boundary is treated as verified; everything that fails it must be recoverable, explained, and audited. This page specifies the stage’s ingress and egress contracts, the HL7 v2 and ASTM E1394 structures it enforces, the asyncio and Pydantic v2 patterns that implement the checks without blocking the pipeline, the tiered error taxonomy that routes every failure, and the property-based tests that prove malformed input never silently corrupts a result.

Context and Pipeline Position

Within the Instrument Data Ingestion & HL7/CSV Pipelines tier, the schema validation and error-handling layer sits between raw acquisition and durable commit. Upstream, the Serial & FTP Polling Architectures capture raw byte streams and delimited files from analyzers and land them on a staging buffer with an immutable tracking identifier. Delimited exports are first reshaped by the CSV to HL7 Transformation stage so that this layer sees a uniform message model regardless of whether the source spoke native HL7 or emitted a flat file. Once a payload passes validation here, it is handed to the async batch processing workers, which own throughput, idempotency, and the transactional LIMS commit. Only after commit does a result become eligible for the Clinical Result Validation & Rule Engine Architecture, which applies reference ranges, delta checks, and critical-value routing. This layer owns none of those clinical verdicts; it owns structural truth.

The separation is deliberate. Because this stage never mutates clinical meaning, its output is deterministic: the same bytes always produce the same verdict, which is what makes replay from the dead-letter queue safe and what lets a regulator reconstruct exactly why a given payload was accepted or rejected months later.

Stage Boundaries

Schema validation exposes one ingress contract and two egress contracts, with explicit failure semantics at every edge. These are the only supported ways a payload crosses the stage.

Ingress — a staged raw payload. The stage accepts an envelope carrying the opaque raw bytes, a tracking_id assigned at acquisition, an instrument_id, a declared wire_format (HL7V2, ASTM_E1394, or CSV), a received_at timestamp, and the transport-level checksum recorded by the polling layer. The stage assumes nothing about internal structure; validity is established here, not trusted from upstream. A payload whose transport checksum does not reconcile is rejected before parsing — a corrupted frame will never become well-formed by re-reading it.

Egress A — a validated canonical result. On success the stage emits a typed, normalized result object: the parsed segments or records coerced into a Pydantic v2 model, with units normalized to UCUM, identifiers resolved to their canonical form, and a validation_trace recording every field-level coercion applied. This object is the only thing the commit stage will accept, and it is immutable once emitted.

Egress B — a quarantine record. When any gate fails, the payload is serialized to a dead-letter queue with full context: the raw bytes, the wire_format, the failing gate, the structured error schema (field path, expected constraint, observed value), the instrument_id, and the tracking_id. Quarantine is terminal for the automated path; a human data engineer or an automated replay job owns re-driving it.

Failure semantics. Transport-checksum reconciliation failures are retryable at the acquisition edge — the file may be re-fetched — but every failure internal to validation (structural, contract, or constraint) is non-retryable and routes straight to quarantine, because deterministic validation of unchanged bytes yields the same verdict. The invariant across every edge: a payload either becomes a fully typed canonical result or a fully contextualized quarantine record. There is no partial, best-effort acceptance — a message with a valid PID but an unparseable OBX is quarantined whole, never committed with a hole where the result should be.

Schema and Protocol Specification

Clinical instruments do not speak one language. This stage enforces distinct structural contracts per wire_format, all converging on the same canonical model. The two that dominate the analyzer floor are HL7 v2.x and ASTM E1394.

HL7 v2 message contract

An HL7 v2.x observation message (ORU^R01) is validated segment by segment. The stage checks segment presence, cardinality, ordering, and the specific fields that a clinical result depends on. Detailed field-level attribution against the site’s canonical dictionary follows the HL7 v2 Segment Mapping contract; the table below is the minimum this stage enforces before a message is eligible to proceed.

Z-segments (site-defined extensions such as ZLR for local result metadata) are validated against a per-instrument custom dictionary versioned alongside the analyzer firmware baseline, so a firmware upgrade that changes a Z-segment layout is caught as a contract violation rather than silently mis-mapped.

ASTM E1394 record sequence

Legacy analyzers frequently emit ASTM E1394 record frames rather than HL7. These are validated by a state machine that enforces record-type ordering, sequence numbers, and per-frame checksums. The full implementation walkthrough lives in validating ASTM 1394 instrument output with Python; the record grammar this stage enforces is summarized below.

Each frame carries a modulo-256 checksum and a monotonically increasing frame sequence number; a checksum mismatch or a sequence gap is a structural failure that quarantines the whole message, because a dropped frame means a lost result and there is no safe way to interpolate one.

Implementation Patterns

Validation runs as a non-blocking stage so that a slow database lookup or a stalled instrument connection never wedges the pipeline. The pattern is a bounded consumer of staged payloads that dispatches each to a validation coroutine under an asyncio.Semaphore, coerces structure with Pydantic v2, and emits either a canonical result or a quarantine record. The concurrency primitives used here are documented in the official Python asyncio library.

Model the canonical result with Pydantic v2 so that structural truth is enforced by construction — an object that cannot be built is a message that must be quarantined.

from __future__ import annotations

from datetime import datetime, timezone
from decimal import Decimal
from enum import Enum

from pydantic import BaseModel, ConfigDict, Field, field_validator


class ValueType(str, Enum):
    NUMERIC = "NM"
    STRING = "ST"
    CODED = "CE"


class Observation(BaseModel):
    """One OBX/R-record observation, coerced to canonical form."""

    model_config = ConfigDict(frozen=True, extra="forbid")

    observation_id: str = Field(min_length=1)   # OBX-3 / test code
    value_type: ValueType                        # OBX-2
    value: Decimal | str                         # OBX-5
    units: str | None = None                     # OBX-6, UCUM-normalized

    @field_validator("units")
    @classmethod
    def _ucum(cls, v: str | None) -> str | None:
        return normalize_ucum(v) if v else v


class CanonicalResult(BaseModel):
    model_config = ConfigDict(frozen=True, extra="forbid")

    tracking_id: str
    instrument_id: str
    patient_id: str            # PID-3 / P-record, non-empty
    accession: str             # OBR-3 filler order number
    service_id: str            # OBR-4 universal service ID
    observations: list[Observation] = Field(min_length=1)
    validated_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))

The validation coroutine wraps parsing and model construction in a timeout guard and converts any structural or contract failure into a typed quarantine outcome rather than letting the exception escape the worker. Exception chaining with raise ... from ... preserves the original stack for forensic triage.

import asyncio


class QuarantineError(Exception):
    """Non-retryable: the payload cannot become a valid CanonicalResult."""

    def __init__(self, gate: str, detail: str) -> None:
        super().__init__(f"{gate}: {detail}")
        self.gate = gate
        self.detail = detail


async def validate_payload(
    envelope: StagedEnvelope,
    sem: asyncio.Semaphore,
    *,
    timeout_s: float = 5.0,
) -> CanonicalResult:
    async with sem:
        try:
            async with asyncio.timeout(timeout_s):
                parsed = await parse_wire(envelope)          # HL7 / ASTM / CSV
                return CanonicalResult.model_validate(parsed)
        except (StructuralParseError, ChecksumError) as exc:
            raise QuarantineError("structural", str(exc)) from exc
        except ValidationError as exc:                        # pydantic
            raise QuarantineError("contract", exc.json()) from exc
        except TimeoutError as exc:
            raise QuarantineError("structural", "validation timeout") from exc

Backpressure is handled by the semaphore and a bounded input queue: when validation cannot keep pace with acquisition, the queue fills and the polling layer stops fetching rather than exhausting memory. Because each coroutine is pure with respect to the input bytes, the pool can be scaled horizontally without any ordering concern — order is re-established, where clinically required, only at the commit stage.

Error Classification and Handling

Every failure is sorted into a three-tier taxonomy, and the tier — not the individual exception — decides the workflow. Collapsing these tiers is the classic defect that turns a transient outage into a wedged queue or lets a malformed frame masquerade as a valid result.

Only the transport tier is retryable, and it is retryable at the acquisition edge — the file is re-fetched — never by re-running validation over identical bytes, because a deterministic validator returns the same verdict every time. Schema and semantic failures are terminal for the automated path; they route to the dead-letter queue carrying the raw payload, the failing gate, the attempt count, and the full raise ... from ... exception chain. Every gate decision — accept, coerce, or quarantine — is written to append-only structured JSON logs carrying the tracking_id as a correlation key, so a single payload can be traced across acquisition, this stage, and commit. Quarantined payloads are never discarded: the dead-letter queue is a durable, replayable store, and a fixed dictionary or an instrument recalibration lets an engineer re-drive a whole cohort of previously rejected messages through the identical validation path.

Regulatory Touchpoints

Because this stage is the boundary where raw instrument output either becomes a trusted clinical record or is held for review, several regulatory clauses bind directly to its behavior. Under CLIA §493.1253, the laboratory must establish and verify performance specifications and ensure that the systems handling results maintain their integrity; the deterministic, versioned validation contract and its append-only decision log are the evidence that no malformed result reaches the report. CAP All Common (COM.30000 / COM.40000) requires documented verification that transmitted data match instrument output and that interface changes are validated before use — satisfied here by binding each Z-segment and ASTM dictionary to a firmware baseline and by recording the payload hash on every canonical result. 21 CFR Part 11 §11.10(a) and (e) require validation of systems to ensure accuracy and a secure, computer-generated, time-stamped audit trail that records operator actions and cannot obscure prior entries; every accept, coercion, and quarantine writes an append-only, attributed audit event. Where the payload carries protected health information, the HIPAA Security Rule §164.312©(1) integrity controls apply to the staged bytes and the emitted result; the encryption-in-transit and least-privilege enforcement is specified in Security & Access Controls, and the tenancy rules that determine which results a validator may process are governed by the CLIA/CAP Data Boundaries.

Testing and Validation

The invariants of this stage — no malformed payload committed, every rejection contextualized, identical bytes yielding identical verdicts — are exactly the properties that hand-picked unit examples miss and property-based tests catch. Use hypothesis to generate adversarial payloads and assert that validation never crashes, never partially accepts, and always classifies.

from hypothesis import given, strategies as st


raw_frames = st.one_of(
    st.binary(min_size=0, max_size=2048),            # arbitrary garbage
    valid_hl7_messages(),                            # well-formed ORU^R01
    hl7_with_dropped_segment(),                      # missing MSH / OBR-4
    astm_with_sequence_gap(),                        # broken frame order
)


@given(payload=raw_frames)
def test_validation_is_total_and_classifying(payload: bytes) -> None:
    envelope = StagedEnvelope(raw=payload, wire_format=detect_format(payload))
    outcome = run_validation(envelope)

    # Total: every input yields exactly one of two terminal outcomes.
    assert isinstance(outcome, (CanonicalResult, QuarantineError))

    # Classifying: a quarantine always names a known gate.
    if isinstance(outcome, QuarantineError):
        assert outcome.gate in {"structural", "contract", "semantic"}

    # Never partial: an accepted result carries at least one observation.
    if isinstance(outcome, CanonicalResult):
        assert len(outcome.observations) >= 1

Two additional test classes round out coverage. A golden-file fixture pins byte-exact HL7 and ASTM payloads with their expected canonical results (or expected quarantine gate), so any change to the parser, the dictionary, or UCUM normalization that shifts a mapping fails the build before it can reach production. A contract test performs a Pydantic round-trip — CanonicalResult.model_validate(result.model_dump()) — asserting that the object the commit stage consumes stays compatible with what this stage emits, so a renamed field breaks the build rather than the pipeline. Determinism itself is tested directly: validating the same bytes twice must produce byte-identical outcomes, which is the property that makes dead-letter-queue replay safe.

Related

• Serial & FTP Polling Architectures — the acquisition layer that stages the raw payloads this stage validates.
• CSV to HL7 Transformation — reshapes delimited exports into the message model this stage checks uniformly.
• Async Batch Processing — the concurrency stage that commits validated results idempotently to the LIMS.
• Validating ASTM 1394 instrument output with Python — the full state-machine parser implementation for legacy ASTM frames.
• HL7 v2 Segment Mapping — the canonical field dictionary this stage enforces messages against.

Part of: Instrument Data Ingestion & HL7/CSV Pipelines.

Frequently Asked Questions