Bulk export and async jobs: serving large rosters with $davinci-data-export

Part 4 of 7 — Roster API Series: migrating member-file ingestion from CSV over SFTP to a FHIR API.

Parts 2 and 3 gave us a versioned Group with explicit add/remove operations. Now the consumer side: someone needs the whole list — members, their Patient records, Coverage, attributed providers. For 200k members that’s hundreds of megabytes. A synchronous GET will time out, and paging through Group.member one reference at a time means 200k follow-up reads.

FHIR already solved this with the Bulk Data Access IG, and ATR profiles it as $davinci-data-export.

The async pattern in one picture

client                                     server
  │  GET Group/123/$davinci-data-export        │
  │  Prefer: respond-async ─────────────────►  │  queue job
  │  ◄──────────── 202 Accepted                │
  │                Content-Location: /jobs/abc │
  │                                            │
  │  GET /jobs/abc ─────────────────────────►  │
  │  ◄──────────── 202  X-Progress: 40%        │  (repeat, honoring Retry-After)
  │                Retry-After: 5              │
  │                                            │
  │  GET /jobs/abc ─────────────────────────►  │
  │  ◄──────────── 200 manifest {output:[…]}   │
  │                                            │
  │  GET /files/abc/Patient.ndjson ─────────►  │  one file per resource type

Interesting detail: this is structurally the same as the SFTP flow — the server writes files, the client picks them up. The difference is that every step has a status code, the file list is explicit, and access is scoped per request.

What ATR adds on top of Bulk Data

From the ATR operation definition and spec:

exportType=hl7.fhir.us.davinci-atr identifies the export as a member attribution export.
Minimum resource types are Group, Patient and Coverage; requests without them are rejected.
patient (0..*) narrows the export to specific members; omitted means all members.
_since / _until bound the export by last update — which is how you get a delta export instead of a full one.
The completion manifest must set requiresAccessToken: true, and the server must check the client may access that Group instance, not just Groups in general.

One inconsistency to know about: the ATR spec examples pass resource types as resourceTypes=... while the requirement text and the base Bulk Data IG use _type. I accept both.

Server: kick-off, status, files

# export.py
import json
import uuid
from datetime import datetime, timezone
from pathlib import Path

from fastapi import BackgroundTasks, FastAPI, Header, HTTPException, Request
from fastapi.responses import FileResponse, JSONResponse, Response

from app import GROUPS, current  # from Part 3

app = FastAPI()
EXPORT_DIR = Path("exports")
JOBS: dict[str, dict] = {}
REQUIRED = {"Group", "Patient", "Coverage"}
RESOURCES: dict[str, dict] = {}  # "Patient/123" -> resource, loaded from Part 2's bundle


def run_export(job_id: str, gid: str, types: set[str], since: str | None):
    job = JOBS[job_id]
    out_dir = EXPORT_DIR / job_id
    out_dir.mkdir(parents=True, exist_ok=True)
    group = current(gid)

    refs = {m["entity"]["reference"] for m in group["member"] if "reference" in m["entity"]}
    refs |= {e["valueReference"]["reference"]
             for m in group["member"] for e in m.get("extension", [])
             if "reference" in e.get("valueReference", {})}

    by_type: dict[str, list[dict]] = {"Group": [group]}
    for ref in refs:
        r = RESOURCES.get(ref)
        if r and r["resourceType"] in types:
            if since and r.get("meta", {}).get("lastUpdated", "") < since:
                continue
            by_type.setdefault(r["resourceType"], []).append(r)

    output = []
    for i, (rtype, rows) in enumerate(by_type.items(), start=1):
        path = out_dir / f"{rtype}.ndjson"
        with path.open("w") as f:
            for r in rows:
                f.write(json.dumps(r, separators=(",", ":")) + "\n")
        output.append({"type": rtype,
                       "url": f"{job['base']}/files/{job_id}/{path.name}",
                       "count": len(rows)})
        job["progress"] = f"{int(100 * i / len(by_type))}%"

    job["manifest"] = {
        "transactionTime": job["started"],
        "request": job["request"],
        "requiresAccessToken": True,
        "output": output,
        "error": [],
    }


@app.get("/Group/{gid}/$davinci-data-export")
def kickoff(gid: str, request: Request, background: BackgroundTasks,
            prefer: str | None = Header(default=None),
            exportType: str | None = None,
            _type: str | None = None,
            resourceTypes: str | None = None,
            _since: str | None = None):
    if prefer != "respond-async":
        raise HTTPException(400, "Prefer: respond-async is required")
    if exportType != "hl7.fhir.us.davinci-atr":
        raise HTTPException(400, "exportType must be hl7.fhir.us.davinci-atr")
    if gid not in GROUPS:
        raise HTTPException(404)

    types = {t.strip() for t in (_type or resourceTypes or "").split(",") if t.strip()}
    types = types or REQUIRED
    if not REQUIRED <= types:
        raise HTTPException(400, f"Export must include {sorted(REQUIRED)}")

    job_id = uuid.uuid4().hex
    base = str(request.base_url).rstrip("/")
    JOBS[job_id] = {"started": datetime.now(timezone.utc).isoformat(),
                    "request": str(request.url), "base": base,
                    "progress": "0%", "manifest": None}
    background.add_task(run_export, job_id, gid, types, _since)
    return Response(status_code=202,
                    headers={"Content-Location": f"{base}/jobs/{job_id}"})


@app.get("/jobs/{job_id}")
def status(job_id: str):
    job = JOBS.get(job_id)
    if not job:
        raise HTTPException(404)
    if job["manifest"] is None:
        return Response(status_code=202,
                        headers={"X-Progress": job["progress"], "Retry-After": "5"})
    return JSONResponse(job["manifest"])


@app.delete("/jobs/{job_id}")
def cancel(job_id: str):
    JOBS.pop(job_id, None)
    return Response(status_code=202)


@app.get("/files/{job_id}/{name}")
def download(job_id: str, name: str):
    path = EXPORT_DIR / job_id / Path(name).name  # no path traversal
    if not path.exists():
        raise HTTPException(404)
    return FileResponse(path, media_type="application/fhir+ndjson")

BackgroundTasks is fine for the demo. In production the kick-off writes a row to a job table and a worker (Celery, RQ, a Cloud Run job — whatever you already run) does the export, so a deploy doesn’t kill in-flight jobs.

Client: kick-off, poll, stream

# client.py
import json
import time

import httpx

BASE = "http://localhost:8000"


def export_group(gid: str, token: str, since: str | None = None) -> dict[str, list[dict]]:
    headers = {"Authorization": f"Bearer {token}",
               "Accept": "application/fhir+json",
               "Prefer": "respond-async"}
    params = {"exportType": "hl7.fhir.us.davinci-atr",
              "_type": "Group,Patient,Coverage"}
    if since:
        params["_since"] = since

    with httpx.Client(base_url=BASE, headers=headers, timeout=30) as c:
        r = c.get(f"/Group/{gid}/$davinci-data-export", params=params)
        r.raise_for_status()
        status_url = r.headers["Content-Location"]

        while True:
            s = c.get(status_url)
            if s.status_code == 200:
                manifest = s.json()
                break
            if s.status_code != 202:
                s.raise_for_status()
            time.sleep(int(s.headers.get("Retry-After", 5)))

        data: dict[str, list[dict]] = {}
        for item in manifest["output"]:
            with c.stream("GET", item["url"]) as resp:
                resp.raise_for_status()
                data[item["type"]] = [json.loads(line) for line in resp.iter_lines() if line]
        return data

Key habits:

Honor Retry-After. Don’t hammer the status endpoint.
Stream NDJSON line by line. One resource per line means you never need the whole file in memory — load straight into a staging table if it’s big.
Store transactionTime. Pass it as _since next run and you get only what changed. That’s the delta feed the CSV world never had.

Why NDJSON

NDJSON is one JSON object per line. That makes files splittable, appendable and loadable by every warehouse (COPY in Snowflake/Redshift, bq load --source_format=NEWLINE_DELIMITED_JSON in BigQuery) without a custom parser. A roster export landing in your warehouse is suddenly the most boring part of the pipeline — which is what you want.

Next: Part 5 — Auth, replacing the long-lived SSH key with SMART Backend Services. Everything above assumed a bearer token; that post is where it comes from.

References

28 Oct 2026 · #fhir #fastapi #python #apis #rosterapiseries

GLB_