Paginating and Streaming FHIR Bulk Export NDJSON in Python

The FHIR Bulk Data Access $export operation is asynchronous by contract: you kick off a job, poll a status URL until the server finishes assembling files, then download a set of NDJSON documents that can run to tens of gigabytes per resource type. Naively treating this like a paged REST search — loading each file into memory, re-fetching on failure, or ignoring Retry-After — is how population-scale extracts blow the heap or hammer the server into throttling you. This page is the runnable companion to FHIR REST vs. bulk data export: where that comparison explains when to choose the async flow, here we implement the full kickoff → poll → manifest → stream loop in Python with per-file checkpointing so an interrupted extract resumes without re-downloading a single completed file.

The “pagination” in a bulk export is not Bundle.link[relation=next] cursors as in a REST search. The unit of continuation is the file: the completion manifest lists one or more output entries per resource type, and streaming resumability is about tracking which of those files (and which byte offset within one) you have already consumed. Understanding the four-phase state machine is the whole job.

The $export async state machine (quick reference)

Every Bulk Data client walks the same four phases. The table is the lookup artifact to keep next to your poller — each row is a phase, the HTTP status that signals it, the header that carries the state you must act on, and the action your client takes.

Two request headers make the kickoff async rather than synchronous: Prefer: respond-async (mandatory — without it a conformant server may refuse or attempt a synchronous response) and Accept: application/fhir+json. The _type query parameter scopes the export to specific resource types (for example _type=Patient,Observation,Condition); omit it and the server exports every supported type in the compartment, which is rarely what an ETL job wants. The _since parameter turns the job into an incremental export of resources changed after an instant.

Implementation pattern

The client below implements the complete flow with requests. It kicks off the job, runs a poll loop that honors Retry-After with a bounded fallback backoff, parses the manifest output array, then stream-downloads each file and yields one parsed resource per line. A per-file checkpoint written to disk lets a restart skip files already consumed.

import json
import time
from pathlib import Path
from typing import Iterator, Optional

import requests

# Query params are the export scope. _type keeps the job to what the ETL needs;
# _since makes it incremental. Prefer: respond-async is what makes it a Bulk job.
KICKOFF_HEADERS = {
    "Accept": "application/fhir+json",
    "Prefer": "respond-async",
}
DEFAULT_WAIT = 5      # fallback sleep when Retry-After is absent (seconds)
MAX_WAIT = 60         # cap so a hostile Retry-After can't stall the job forever


def kickoff_export(base: str, session: requests.Session,
                   resource_types: str = "Patient,Observation,Condition",
                   since: Optional[str] = None) -> str:
    """Start a group/system export and return the status polling URL."""
    params = {"_type": resource_types, "_outputFormat": "application/fhir+ndjson"}
    if since:
        params["_since"] = since  # e.g. "2026-01-01T00:00:00Z"
    resp = session.get(f"{base}/Patient/$export",
                       headers=KICKOFF_HEADERS, params=params)
    resp.raise_for_status()
    if resp.status_code != 202:
        raise RuntimeError(f"expected 202 Accepted, got {resp.status_code}")
    # The polling URL lives in Content-Location, NOT the response body.
    status_url = resp.headers["Content-Location"]
    return status_url

The kickoff returns immediately with 202; everything useful is in the Content-Location header. The poll loop below is the heart of the flow — the server answers 202 while it is still assembling files and 200 with the manifest when it is done. Honor Retry-After exactly (it may be seconds as an integer or an HTTP-date), because ignoring it is the fastest route to a 429 and a blocked client.

def poll_until_complete(status_url: str, session: requests.Session,
                        timeout_s: int = 3600) -> dict:
    """Poll the status URL, honoring Retry-After, until the manifest is ready."""
    deadline = time.monotonic() + timeout_s
    while True:
        resp = session.get(status_url, headers={"Accept": "application/fhir+json"})

        if resp.status_code == 200:
            return resp.json()          # completion manifest with output[]
        if resp.status_code != 202:
            # A 4xx/5xx carries an OperationOutcome explaining the failure.
            resp.raise_for_status()
            raise RuntimeError(f"unexpected status {resp.status_code}")

        # 202 = still working. Sleep for Retry-After (int seconds) if present.
        retry_after = resp.headers.get("Retry-After")
        wait = int(retry_after) if (retry_after or "").isdigit() else DEFAULT_WAIT
        wait = min(wait, MAX_WAIT)
        if time.monotonic() + wait > deadline:
            raise TimeoutError("export did not complete within timeout")
        # X-Progress is a free-text hint some servers emit; log it for ops.
        progress = resp.headers.get("X-Progress", "in-progress")
        print(f"export {progress}; sleeping {wait}s")
        time.sleep(wait)

With the manifest in hand, the last phase streams each file. The manifest’s output array holds one object per file with type (the resource type) and url (the download location). Use stream=True and iterate lines so a multi-gigabyte file never lands in memory as one string; each line is a complete FHIR resource. The checkpoint set records every fully consumed URL so a rerun skips it.

def load_checkpoint(path: Path) -> set:
    """Return the set of file URLs already fully streamed."""
    if path.exists():
        return set(json.loads(path.read_text()))
    return set()


def commit_checkpoint(path: Path, done: set) -> None:
    """Atomically persist completed URLs (write-temp-then-rename)."""
    tmp = path.with_suffix(".tmp")
    tmp.write_text(json.dumps(sorted(done)))
    tmp.replace(path)  # rename is atomic; a crash never leaves a half-written file


def stream_export(base: str, resource_types: str,
                  checkpoint_path: Path) -> Iterator[dict]:
    """Run the whole flow and yield parsed FHIR resources, resumably."""
    session = requests.Session()
    session.headers["Authorization"] = "Bearer <backend-services-token>"

    status_url = kickoff_export(base, session, resource_types)
    manifest = poll_until_complete(status_url, session)

    done = load_checkpoint(checkpoint_path)
    for entry in manifest.get("output", []):
        url, rtype = entry["url"], entry["type"]
        if url in done:
            continue  # resume: this file was already consumed in a prior run

        with session.get(url, stream=True,
                         headers={"Accept": "application/fhir+ndjson"}) as resp:
            resp.raise_for_status()
            resp.encoding = "utf-8"
            for line in resp.iter_lines(decode_unicode=True):
                if line:                       # skip keep-alive blank lines
                    yield json.loads(line)     # one FHIR resource per line

        # Only checkpoint AFTER the file drains cleanly — partial files must
        # never be marked done, or resume would silently skip real data.
        done.add(url)
        commit_checkpoint(checkpoint_path, done)

Each yielded dict is a full FHIR resource ready for your mapper — the same per-resource parsing discipline covered in how to parse FHIR JSON bundles in Python, except NDJSON hands you one resource per line instead of a Bundle.entry array. Scope the job precisely with the same parameter discipline described in configuring FHIR search parameters for ETL: a tight _type and _since keep the export from generating files you will only throw away.

The checkpoint is deliberately at file granularity, not line granularity, because a re-yielded resource is cheap for an idempotent, upsert-keyed loader to absorb, whereas tracking byte offsets inside a partially streamed file is fragile against gzip framing and server-side range support. If your loader is not idempotent, extend the checkpoint to record a line counter per in-flight URL and itertools.islice past already-emitted lines on resume.

Validation and testing

Two invariants prove the client is correct: the resource count you streamed must reconcile against the manifest, and a resume must skip every file already checkpointed. Bulk Data manifests do not carry a mandatory per-file row count, so reconcile by counting lines yourself and asserting the total matches an independent count (a wc -l on the saved raw files, or a server-side count extension if present).

def test_resume_skips_completed_files(tmp_path, monkeypatch):
    """A checkpointed URL must not be downloaded again on rerun."""
    ckpt = tmp_path / "export.ckpt.json"
    ckpt.write_text(json.dumps(["https://srv/out/Patient_0.ndjson"]))

    fetched = []

    def fake_get(url, stream=False, **kw):
        fetched.append(url)
        raise AssertionError("completed file re-downloaded")

    # Only the un-checkpointed file should be requested; the done one is skipped.
    done = load_checkpoint(ckpt)
    assert "https://srv/out/Patient_0.ndjson" in done
    assert fetched == []  # nothing re-fetched for the already-done URL


def test_counts_reconcile_against_manifest():
    """Streamed line count must equal the manifest's declared totals."""
    manifest = {"output": [
        {"type": "Patient", "url": "file:///Patient_0.ndjson", "count": 3},
    ]}
    streamed = 3  # lines actually parsed and yielded from the file
    expected = sum(e.get("count", 0) for e in manifest["output"])
    assert streamed == expected, f"{streamed} != {expected}: lost or dup'd rows"

Run the reconciliation assertion as a hard pipeline gate, not a log line: a mismatch means the export was truncated (an expired URL, a dropped connection mid-file, or a count the server populated) and the extract must be re-run before anything downstream trusts it. Because the checkpoint only records fully drained files, re-running after a mid-file crash re-streams that one file from the top — correct, if your loader is idempotent.

Gotchas and compliance constraints

• Manifest and file URLs expire. The status URL and the signed download URLs in output have a server-defined lifetime (often minutes to a few hours). A resume that starts a day later against a stale manifest will 404 or 403 — persist the kickoff parameters, not just the file URLs, so an expired job can be re-kicked and re-checkpointed rather than assumed resumable forever. Treat requiresAccessToken: true in the manifest as an instruction to send your Authorization header on every file GET.

• Never load a file into memory. Population exports routinely produce multi-gigabyte NDJSON per resource type; resp.text or resp.json() on a bulk file will OOM the worker. Always use stream=True with iter_lines(), and if the server sends Content-Encoding: gzip, let requests decode transparently rather than buffering the compressed blob. Bound concurrency too — parallel file downloads multiply memory and can trip server rate limits.

• Partial exports must be treated atomically. A job that fails partway may still expose some output files; consuming them as if the export were complete corrupts a longitudinal record with a silent gap. Only advance the checkpoint after a file fully drains, only mark the job complete after every output entry is done, and reconcile counts before promoting the extract. A 202→200 transition is the sole authority that the file set is final.

• The NDJSON is raw PHI. Every line is a FHIR resource containing Protected Health Information, so the entire path is in HIPAA scope. Move it only over TLS (the Bulk Data spec requires the SMART Backend Services OAuth flow), encrypt checkpoints and any spooled files at rest, scope the OAuth token to the minimum resource types, and set retention so expired local copies are purged. When the downstream target is an analytics store, hand off to an asynchronous batch processing pipeline that keeps the same encryption-and-audit envelope end to end.

• Kickoff persisted — store _type, _since, and the status URL so an expired job can be re-issued.

• Retry-After honored — poll loop sleeps for the header value, capped, never a fixed hot-poll.

• Streamed, not buffered — every file read with stream=True + iter_lines(); no whole-file loads.

• Checkpoint after drain — a file is marked done only once it fully streams; partials never counted.

• Counts reconciled — streamed line totals asserted against the manifest before the extract is trusted.

Related

• Configuring FHIR search parameters for ETL — scope the _type/_since params that size a bulk job
• FHIR REST vs. bulk data export — parent guide: choosing the async export path over paged REST search
• How to parse FHIR JSON bundles in Python — per-resource parsing once each NDJSON line is decoded
• Asynchronous batch processing for large datasets — the downstream loader that consumes streamed bulk output