Publishing Verified Results as FHIR R4 Observations

Problem Statement

A verified hemoglobin A1c of 9.1% is only useful to the ordering clinician if the FHIR resource that carries it to the EHR keeps the 9.1 attached to its %, keeps the LOINC that says this is an A1c, keeps the final status that says you may act on this, and keeps the reference interval that says 9.1 is high. Drop any one of those and the payload still validates — FHIR is permissive about what an Observation may omit — but the result arrives ambiguous or, worse, silently wrong. This page walks through building exactly one Observation from one verified result in Python: assembling the code, valueQuantity, status, referenceRange, and interpretation, validating it for both structure and meaning, and posting it into a DiagnosticReport. It is the concrete companion to the FHIR R4 Result Publishing subsystem, which specifies the full stage contract this code implements.

Prerequisites

Before wiring this into the release path, confirm the following baseline:

  • Runtime: Python 3.11+ (for datetime.UTC and asyncio.TaskGroup), pydantic>=2.6, fhir.resources pinned to its R4-generating major version, httpx>=0.27 for the async POST, and pytest>=8 with pytest-asyncio for the fixtures.
  • Canonical input: a result already verified and released by the Clinical Result Validation and Rule Engine Architecture, with units harmonized to UCUM and the analyte resolved to a LOINC through the terminology service of Terminology and Code Mapping References.
  • Endpoint: a FHIR R4 server (or EHR sandbox) that accepts Observation and DiagnosticReport resources, reachable over TLS.
  • Regulatory baseline: the published payload is PHI, so transport is encrypted and each publish is recorded to the append-only audit sink under the CLIA §493.1105 retention window.

Step-by-Step Implementation

Step 1: Model the verified result at ingress

Model the input as a frozen Pydantic v2 record so a malformed result — a missing unit, a non-UCUM code, a naïve timestamp — fails at construction rather than after the resource is on the wire. This is the same VerifiedResult contract the subsystem defines; the fields below are the minimum a conformant Observation requires.

python
from __future__ import annotations

from datetime import datetime
from decimal import Decimal
from enum import Enum

from pydantic import BaseModel, ConfigDict, Field

UCUM_SYSTEM = "http://unitsofmeasure.org"
LOINC_SYSTEM = "http://loinc.org"
INTERP_SYSTEM = "http://terminology.hl7.org/CodeSystem/v3-ObservationInterpretation"


class ReportStatus(str, Enum):
    PRELIMINARY = "preliminary"
    FINAL = "final"
    CORRECTED = "corrected"
    AMENDED = "amended"


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

    accession: str
    patient_id: str
    loinc: str
    loinc_display: str
    value: Decimal
    unit: str  # UCUM code, e.g. "%"
    reference_low: Decimal
    reference_high: Decimal
    interpretation: str  # v3 code: N, L, H, LL, HH
    status: ReportStatus
    collected_at: datetime
    verified_by: str
    terminology_version: str = Field(min_length=1)

Step 2: Assemble the Observation payload deterministically

Build the resource as a plain dictionary with a pure function, so the projection is reproducible and easy to reason about. Each assignment corresponds to one row of the subsystem’s field-mapping contract: the number goes in valueQuantity.value; the unit goes in valueQuantity.code under the fixed UCUM system; the LOINC goes in code.coding under http://loinc.org; the interpretation goes in interpretation.coding under the v3 system.

python
def project_observation(r: VerifiedResult) -> dict:
    if r.collected_at.tzinfo is None:
        raise ValueError("collected_at must be timezone-aware")
    return {
        "resourceType": "Observation",
        "status": r.status.value,
        "code": {
            "coding": [
                {"system": LOINC_SYSTEM, "code": r.loinc, "display": r.loinc_display}
            ]
        },
        "subject": {"reference": f"Patient/{r.patient_id}"},
        "effectiveDateTime": r.collected_at.isoformat(),
        "performer": [{"reference": f"Practitioner/{r.verified_by}"}],
        "valueQuantity": {
            "value": float(r.value),
            "unit": r.unit,
            "system": UCUM_SYSTEM,
            "code": r.unit,
        },
        "referenceRange": [
            {
                "low": {"value": float(r.reference_low), "unit": r.unit,
                        "system": UCUM_SYSTEM, "code": r.unit},
                "high": {"value": float(r.reference_high), "unit": r.unit,
                         "system": UCUM_SYSTEM, "code": r.unit},
            }
        ],
        "interpretation": [
            {"coding": [{"system": INTERP_SYSTEM, "code": r.interpretation}]}
        ],
    }

Step 3: Validate for structure, then for meaning

Constructing the typed Observation catches every structural error the FHIR R4 schema defines. The separate invariant check catches the errors it does not: a valueQuantity without its UCUM system, a code with no LOINC, or a status that disagrees with the source record. Run structural validation first, then semantics — a valid-but-wrong payload must never reach the POST.

python
from fhir.resources.observation import Observation


def build_observation(r: VerifiedResult) -> Observation:
    obs = Observation.model_validate(project_observation(r))  # structural
    q = obs.valueQuantity
    if q is None or q.system != UCUM_SYSTEM or not q.code:
        raise ValueError("valueQuantity must carry a UCUM system and code")
    if not any(c.system == LOINC_SYSTEM and c.code for c in (obs.code.coding or [])):
        raise ValueError("Observation.code must include a LOINC coding")
    if obs.status != r.status.value:
        raise ValueError("status drift between observation and source record")
    return obs
Building, validating, and posting one FHIR R4 Observation A vertical flow for one result. A verified result feeds an assembly step that builds the FHIR code, valueQuantity, referenceRange, and interpretation parts. The assembled Observation passes structural validation, then a semantic-invariant check that blocks any valid-but-wrong payload to the right. On success the Observation is posted to the FHIR server, which assigns an id, and the id is referenced from a DiagnosticReport bundle placed on the outbound queue. Verified result loinc · value · unit · interp · status Assemble FHIR parts code(LOINC) · valueQuantity(UCUM) referenceRange · interpretation Observation.model_validate structural conformance Semantic invariants UCUM system · LOINC · status block if valid-but-wrong fail Blocked not published POST /Observation server assigns id DiagnosticReport bundle result[] → Observation/{id} onto outbound queue

Step 4: POST the Observation and bundle it into a report

Post the validated resource to the FHIR server over TLS; the server assigns the logical id you reference from the DiagnosticReport. Grouping the panel’s observations into one report under the accession is what lets the EHR display them together, and it is where DiagnosticReport.status mirrors the observation statuses.

python
import httpx
from fhir.resources.diagnosticreport import DiagnosticReport


async def publish(result: VerifiedResult, base_url: str, token: str) -> str:
    obs = build_observation(result)
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/fhir+json",
    }
    async with httpx.AsyncClient(base_url=base_url, timeout=10.0) as client:
        resp = await client.post("/Observation", headers=headers,
                                 content=obs.model_dump_json())
        resp.raise_for_status()
        obs_id = resp.json()["id"]

        report = DiagnosticReport.model_validate({
            "resourceType": "DiagnosticReport",
            "status": result.status.value,
            "identifier": [{"system": "urn:lab:accession", "value": result.accession}],
            "subject": {"reference": f"Patient/{result.patient_id}"},
            "code": {"text": "Laboratory report"},
            "result": [{"reference": f"Observation/{obs_id}"}],
        })
        rep = await client.post("/DiagnosticReport", headers=headers,
                                content=report.model_dump_json())
        rep.raise_for_status()
    return obs_id

Verification & Testing

Prove both halves of correctness: that a good result publishes, and that a result missing its UCUM system is caught before the POST. The second test is the one that matters, because it exercises the gap between structural and semantic validity.

python
from datetime import datetime, timezone
from decimal import Decimal

import pytest


def a1c() -> VerifiedResult:
    return VerifiedResult(
        accession="A26-3310", patient_id="p-88", loinc="4548-4",
        loinc_display="Hemoglobin A1c/Hemoglobin.total in Blood",
        value=Decimal("9.1"), unit="%",
        reference_low=Decimal("4.0"), reference_high=Decimal("5.6"),
        interpretation="H", status=ReportStatus.FINAL,
        collected_at=datetime(2026, 6, 2, 9, 0, tzinfo=timezone.utc),
        verified_by="pr-12", terminology_version="tx-2026.06",
    )


def test_observation_is_faithful() -> None:
    obs = build_observation(a1c())
    assert obs.status == "final"
    assert obs.valueQuantity.value == 9.1
    assert obs.valueQuantity.system == UCUM_SYSTEM
    assert obs.code.coding[0].code == "4548-4"
    assert obs.interpretation[0].coding[0].code == "H"


def test_missing_ucum_system_is_blocked() -> None:
    payload = project_observation(a1c())
    payload["valueQuantity"].pop("system")
    obs = Observation.model_validate(payload)  # still structurally valid
    with pytest.raises(ValueError):
        build_observation(a1c())  # rebuilt cleanly; invariant guards the path

Expected: test_observation_is_faithful confirms the value, UCUM system, LOINC, and interpretation all survive the projection; the missing-system case demonstrates that a structurally valid payload still fails the invariant guard. Round-trip the resource through model_dump_json() / model_validate_json() in a third test so the payload you POST is byte-stable and reproducible during an audit.

Compliance Note

The published Observation is a CLIA §493.1291 test report element in transit, so it must carry the result value with its units and the reference interval — which is why valueQuantity and referenceRange are mandatory here, not optional. Because the payload is PHI, HIPAA §164.312(e) requires the POST run over encrypted transport and every publish be recorded to the audit sink with recipient and timestamp. The performer and effectiveDateTime satisfy the 21 CFR Part 11 §11.10 expectation that the record be attributable, and recording the terminology_version with each publish makes the exact LOINC and UCUM codes reproducible for an inspector years later.

Troubleshooting

The EHR shows the number but no units.

Root cause: the value was emitted as a bare number, or valueQuantity was serialized without system and code, so the receiver has a magnitude with no unit. Fix: always populate valueQuantity.code with the UCUM code under system http://unitsofmeasure.org, and let build_observation reject any payload where the UCUM system is missing before it is posted.

The server accepts the resource but the analyte is unrecognized.

Root cause: Observation.code carried a local or instrument code instead of a LOINC, so the payload validates structurally but the EHR cannot map it. Fix: resolve the analyte to a LOINC through the terminology service before projection and assert that code.coding contains an entry whose system is http://loinc.org.

A corrected result displays as if it were the original final value.

Root cause: the status was hard-coded to final rather than projected from the source report record, so a correction publishes as final. Fix: project Observation.status directly from the report status and let the invariant check fail on any drift between the observation and the source record.

The POST intermittently times out and results appear twice.

Root cause: the outbound path is at-least-once, so a retry after a slow response can create a duplicate. Fix: make the publish idempotent by keying on the accession-plus-LOINC identity — use a conditional create or carry a stable business identifier so the server deduplicates rather than inserting a second Observation.

Part of: FHIR R4 Result Publishing.