fetching-fhir-resources

Fetching FHIR R4 Resources for OpenMed

FHIR R4 is the modern EHR API: a RESTful, JSON-or-XML interface over resources like Patient, Encounter, Condition, Observation, DiagnosticReport, and DocumentReference. The unstructured clinical text you want for NLP lives in DocumentReference.content.attachment and DiagnosticReport.presentedForm — usually base64-encoded PDF, RTF, or plain text. This skill pulls those resources, pages through results, decodes the attachments, and hands the narrative to OpenMed.

When to use

• You have FHIR R4 access to an EHR (Epic, Oracle Health/Cerner, HAPI, Medplum, Azure/Google/AWS HealthLake) and want note text for de-id and NER.
• You need to page a large search result set safely (Bundle.link[next]).
• You want to pull a patient's documents/reports and rejoin NLP output by patient and encounter.

FHIR REST in one minute

Search is GET [base]/[Type]?param=value. Results come back as a searchset Bundle; the next page is the URL in Bundle.link where relation == "next". Use _count to size pages, _revinclude to pull related resources in one round trip, and _since/_lastUpdated for incremental sync.

GET /Patient?identifier=http://hospital.org/mrn|12345
GET /DocumentReference?patient=Patient/abc&category=clinical-note&_count=50
GET /DiagnosticReport?patient=Patient/abc&_revinclude=Observation:related

Quick start

Page a search, decode attachments, hand narrative to OpenMed:

import base64
import requests
import openmed

BASE = "https://fhir.example.org/r4"
HEADERS = {"Accept": "application/fhir+json", "Authorization": "Bearer <token>"}

def iter_bundle(url, params=None):
    """Yield resources across all pages following Bundle.link[next]."""
    while url:
        bundle = requests.get(url, params=params, headers=HEADERS, timeout=30).json()
        for entry in bundle.get("entry", []):
            yield entry.get("resource", {})
        params = None  # next links are fully-qualified
        url = next(
            (l["url"] for l in bundle.get("link", []) if l.get("relation") == "next"),
            None,
        )

def attachment_text(att):
    """Decode a FHIR Attachment to text (handles base64 and inline text/plain)."""
    if att.get("data"):
        raw = base64.b64decode(att["data"])
        if att.get("contentType", "").startswith("text/"):
            return raw.decode("utf-8", "replace")
        return ""  # PDF/RTF: route to OpenMed multimodal/OCR intake instead
    return ""

# Pull a patient's clinical notes and analyze each.
for doc in iter_bundle(f"{BASE}/DocumentReference",
                       {"patient": "Patient/abc",
                        "category": "clinical-note", "_count": 50}):
    for content in doc.get("content", []):
        text = attachment_text(content.get("attachment", {}))
        if not text.strip():
            continue
        deid = openmed.deidentify(text, method="replace", policy="hipaa_safe_harbor")
        result = openmed.analyze_text(deid.text, output_format="dict")
        patient_ref = doc.get("subject", {}).get("reference")  # rejoin key