Using fhir.resources for Python ETL

Production-grade clinical pipelines demand deterministic parsing, strict schema validation, and auditable transformation logic. The fhir.resources package gives you a Pydantic v2-backed, strongly typed interface to the HL7 FHIR R4/R5 specification, so engineering teams can replace fragile dictionary manipulation with contract-driven transforms. Within the Clinical Data Parsing & Transformation Workflows pipeline, this page covers how to make fhir.resources the validation boundary of a Python extract-transform-load job: enforcing structural integrity at ingestion while preserving clinical semantics for downstream analytics, machine-learning feature stores, and regulatory reporting. The goal is not “parse some JSON” — it is to treat every resource instantiation as a typed contract whose failures are observable, recoverable, and compliant.

Prerequisites & Context

Before wiring fhir.resources into a pipeline, confirm the following are in place:

• Python 3.10+ with fhir.resources>=7.0 (Pydantic v2 era) and pydantic>=2.5 pinned in a lockfile — the v1→v2 API break changed every model method name.
• A source of FHIR payloads: a FHIR server endpoint, a bulk NDJSON export, or staged JSON from an upstream converter.
• If you ingest legacy HL7 v2 feeds, a deterministic v2-to-FHIR translation layer running ahead of this stage — see the HL7 Python Library Integration Guide for the segment-to-resource mapping that must complete before any model_validate call.
• A dead-letter queue (DLQ) or quarantine sink (Kafka topic, SQS queue, or an encrypted object-store prefix) for records that fail validation.
• Familiarity with the FHIR resource hierarchy so you know which resource types and required elements you are validating against.

If you are reading raw Bundle payloads off the wire, pair this page with how to parse FHIR JSON bundles in Python for the bundle-walking patterns that feed the validators below.

Concept & Spec Detail: fhir.resources as a Typed Contract

fhir.resources generates one Pydantic model per FHIR resource type directly from the published StructureDefinitions. When you call Patient.model_validate(data), Pydantic enforces FHIR cardinality, required elements, primitive regex constraints, and choice-type ([x]) exclusivity at instantiation time. Unlike loose json.loads deserialization, it raises pydantic.ValidationError on any structural deviation — which is exactly the signal an idempotent pipeline needs to branch on.

The methods you will use most, and how they differ, are worth keeping straight:

Two behaviours matter for correctness. First, exclude_unset=True preserves the distinction between “field absent” and “field explicitly null” — critical for clinical null semantics, where an absent lab value and a value that was asked but unknown must not collapse into the same output. Second, FHIR primitives (date, dateTime, decimal, code, uri) are strict: a partial date like 2023-05, a timezone-naive timestamp, or a locale-formatted decimal will be rejected unless normalized first. Treat fhir.resources as the contract, not the janitor — clean inputs in a stage that runs before validation.

Implementation

Step 1 — Validate at the ingestion boundary with DLQ fallback

Wrap every instantiation in explicit exception handling. Structural failures route to a DLQ with enough context to triage, but without persisting raw clinical payloads to unencrypted log targets. Use a stable cryptographic hash of the canonicalized payload as the correlation key — never Python’s built-in hash(), which is salted per-process and not reproducible across runs.

import json
import hashlib
import logging
from pydantic import ValidationError
from fhir.resources.patient import Patient
from fhir.resources.observation import Observation

logger = logging.getLogger("clinical.etl.validation")

# Dispatch table keyed by FHIR resourceType -> model class.
VALIDATORS = {
    "Patient": Patient,
    "Observation": Observation,
    # Extend with additional resource types as the pipeline grows.
}

def ingest_fhir_bundle(payload: dict) -> list[dict]:
    """Validate and route FHIR resources from a Bundle, with DLQ fallback."""
    valid_resources: list[dict] = []
    for entry in payload.get("entry", []):
        resource_data = entry.get("resource", {})
        resource_type = resource_data.get("resourceType")
        model = VALIDATORS.get(resource_type)

        try:
            if model is not None:
                model.model_validate(resource_data)  # raises on any FHIR violation
            valid_resources.append(resource_data)
        except ValidationError as exc:
            dlq_record = {
                "error": "validation_failure",
                "resource_type": resource_type,
                # Stable, reproducible correlation key — NOT Python's hash().
                "payload_hash": hashlib.sha256(
                    json.dumps(resource_data, sort_keys=True).encode()
                ).hexdigest()[:16],
                "details": str(exc),
            }
            logger.error("DLQ routed: %s", dlq_record)
            # Push dlq_record to the encrypted DLQ topic in production.
    return valid_resources

The DLQ record above deliberately omits the raw payload. Persisting PHI to a log sink is the single most common HIPAA violation in clinical ETL; carry the payload_hash instead and resolve the original from the encrypted quarantine store only when a human triages it.

Validation check:

# A malformed Patient (gender is a bound code; "M" is not a valid value) must be quarantined.
out = ingest_fhir_bundle({"entry": [{"resource": {"resourceType": "Patient", "gender": "M"}}]})
assert out == [], "invalid gender code should have routed to the DLQ, not passed through"

Step 2 — Normalize clinical primitives before validation

Real EHR exports routinely carry partial dates, timezone-naive timestamps, and over-precise decimals. Run a normalization pass first so strict FHIR primitives validate cleanly. The deeper conversion rules for each data type live in Type Coercion for Clinical Data Types; this stage applies the minimum needed to satisfy the contract.

import re
from datetime import datetime, timezone
from decimal import Decimal, InvalidOperation

def normalize_fhir_primitives(data: dict) -> dict:
    """Pre-validate normalization for common clinical primitives."""
    # Expand partial dates: "2023" -> "2023-01-01", "2023-05" -> "2023-05-01".
    if "birthDate" in data and isinstance(data["birthDate"], str):
        bd = data["birthDate"]
        if re.fullmatch(r"\d{4}", bd):
            data["birthDate"] = bd + "-01-01"
        elif re.fullmatch(r"\d{4}-\d{2}", bd):
            data["birthDate"] = bd + "-01"

    # Label timezone-naive effectiveDateTime explicitly. If the source emits
    # local time, use .astimezone() instead of assuming UTC.
    if "effectiveDateTime" in data and isinstance(data["effectiveDateTime"], str):
        dt_str = data["effectiveDateTime"]
        if not dt_str.endswith("Z") and not re.search(r"[+-]\d{2}:?\d{2}$", dt_str):
            dt = datetime.fromisoformat(dt_str).replace(tzinfo=timezone.utc)
            data["effectiveDateTime"] = dt.isoformat()

    # Enforce decimal precision for clinical measurements without losing the value.
    if "valueQuantity" in data and "value" in data["valueQuantity"]:
        try:
            val = Decimal(str(data["valueQuantity"]["value"]))
            # Stored as float for FHIR JSON; track significant figures separately.
            data["valueQuantity"]["value"] = float(val.quantize(Decimal("0.001")))
        except InvalidOperation:
            raise ValueError("Non-numeric clinical measurement detected")

    return data

Validation check:

assert normalize_fhir_primitives({"birthDate": "1980"})["birthDate"] == "1980-01-01"

Step 3 — Map legacy null semantics to data-absent-reason

FHIR does not carry the HL7 v2/v3 nullFlavor attribute natively. To preserve the difference between not asked, asked but unknown, and masked, map each flavor onto the FHIR data-absent-reason extension. Note the value type: the R4 extension definition specifies valueCode, not valueCodeableConcept.

# HL7 v2/v3 nullFlavor -> FHIR data-absent-reason codes.
NULL_FLAVOR_TO_DAR = {
    "UNK":  "unknown",
    "ASKU": "asked-unknown",
    "NASK": "not-asked",
    "MSK":  "masked",
}

DATA_ABSENT_REASON_URL = "http://hl7.org/fhir/StructureDefinition/data-absent-reason"

def apply_data_absent_reason(resource: dict, null_flavor: str) -> dict:
    """Attach a FHIR-compliant data-absent-reason extension for a legacy null."""
    dar_code = NULL_FLAVOR_TO_DAR.get(null_flavor)
    if not dar_code:
        return resource  # Unknown flavor: leave unchanged and log upstream.

    resource.setdefault("extension", []).append({
        "url": DATA_ABSENT_REASON_URL,
        "valueCode": dar_code,
    })
    return resource

The full mapping matrix — including how to attach the extension at element level versus resource level — is detailed in Handling nullFlavor in FHIR resource extraction.

Step 4 — Stream and serialize at scale

fhir.resources models are memory-intensive when instantiated in bulk, so never load an entire export into a list. Process in generator-based chunks and serialize validated resources to NDJSON for downstream sinks.

from typing import Iterator, Generator

def chunked_fhir_processing(
    source_iter: Iterator[dict],
    chunk_size: int = 5000,
) -> Generator[list[dict], None, None]:
    """Stream FHIR resources in memory-safe chunks."""
    buffer: list[dict] = []
    for resource in source_iter:
        buffer.append(resource)
        if len(buffer) >= chunk_size:
            yield buffer
            buffer = []
    if buffer:
        yield buffer

def batch_validate_and_dump(chunk: list[dict]) -> list[str]:
    """Validate a chunk and serialize survivors to NDJSON lines."""
    serialized: list[str] = []
    for res in chunk:
        try:
            if res.get("resourceType") == "Patient":
                validated = Patient.model_validate(res)
                serialized.append(validated.model_dump_json(exclude_unset=True))
        except ValidationError:
            continue  # Route to DLQ in production; skipped here for brevity.
    return serialized

For the DataFrame-oriented path — PyArrow-backed schemas, memory-mapped I/O, and chunked model_dump() into columnar form — see Optimizing pandas for FHIR JSON parsing. For multi-worker pipelines that fan these chunks across a queue, the backpressure and idempotency patterns in async batch processing for large datasets apply directly.

Edge Cases & Vendor Deviations

Even well-formed exports from major EHRs carry quirks that fhir.resources will reject because it follows the base spec strictly. Handle these in the normalization stage, not by loosening validation.