import argparse import json import time import uuid from typing import Iterable import jwt import requests from cryptography.hazmat.primitives import serialization # ========================= # CONFIG (edit these) # ========================= CLIENT_ID = "1fed7724-fddf-469e-8b02-d11f338e264c" TOKEN_URL = "https://fhir.epic.com/interconnect-fhir-oauth/oauth2/token" FHIR_BASE = "https://fhir.epic.com/interconnect-fhir-oauth/api/FHIR/R4" KID = "jwks_live_patients.json" # must match your JWKS registration 'kid' PRIVATE_KEY_PATH = "keys/privatekey.pem" # ========================= # AUTH (same pattern as yours) # ========================= with open(PRIVATE_KEY_PATH, "rb") as fp: KEY = serialization.load_pem_private_key(fp.read(), password=None) def client_assertion() -> str: now = int(time.time()) payload = { "iss": CLIENT_ID, "sub": CLIENT_ID, "aud": TOKEN_URL, "jti": str(uuid.uuid4()), "iat": now, "nbf": now, "exp": now + 300, } headers = {"kid": KID} return jwt.encode(payload, KEY, algorithm="RS384", headers=headers) def get_token() -> dict: assertion = client_assertion() r = requests.post( TOKEN_URL, data={ "grant_type": "client_credentials", "client_assertion_type": "urn:ietf:params:oauth:client-assertion-type:jwt-bearer", "client_assertion": assertion, }, timeout=15, ) r.raise_for_status() return r.json() def headers(access_token: str) -> dict: return { "Authorization": f"Bearer {access_token}", "Accept": "application/fhir+json", } # ========================= # UTIL # ========================= def bundle_entries(bundle: dict) -> list[dict]: return bundle.get("entry") or [] def get_next_link(bundle: dict) -> str | None: for link in bundle.get("link") or []: if link.get("relation") == "next": return link.get("url") return None def is_operation_outcome(obj: dict) -> bool: return isinstance(obj, dict) and obj.get("resourceType") == "OperationOutcome" def write_patients(out_path: str, patients: Iterable[dict], ndjson: bool) -> None: if ndjson: with open(out_path, "w", encoding="utf-8") as f: for p in patients: f.write(json.dumps(p)) f.write("\n") else: with open(out_path, "w", encoding="utf-8") as f: json.dump(list(patients), f, indent=2) # ========================= # MODE A: Fetch "all patients" from a List # GET /List?code=patients&identifier=... # ========================= def fetch_list_bundle(access_token: str, code: str, identifier: str) -> dict: url = f"{FHIR_BASE}/List" params = {"code": code, "identifier": identifier} r = requests.get(url, headers=headers(access_token), params=params, timeout=30) r.raise_for_status() return r.json() def patient_refs_from_list_bundle(list_bundle: dict) -> list[str]: """ Returns references like 'Patient/abc123' """ entries = bundle_entries(list_bundle) if not entries: return [] # take first List resource in bundle list_res = entries[0].get("resource") or {} list_entries = list_res.get("entry") or [] refs: list[str] = [] for e in list_entries: ref = ((e.get("item") or {}).get("reference")) if ref: refs.append(ref) return refs def fetch_patient_by_ref(access_token: str, ref: str) -> dict: """ ref may be: - 'Patient/{id}' - full URL """ if ref.startswith("http://") or ref.startswith("https://"): url = ref else: # normalize: if ref is "Patient/xyz", build absolute URL url = f"{FHIR_BASE}/{ref.lstrip('/')}" r = requests.get(url, headers=headers(access_token), timeout=30) r.raise_for_status() return r.json() def fetch_all_patients_from_list(access_token: str, code: str, identifier: str) -> list[dict]: list_bundle = fetch_list_bundle(access_token, code, identifier) refs = patient_refs_from_list_bundle(list_bundle) patients: list[dict] = [] for i, ref in enumerate(refs, start=1): print(f"[{i}/{len(refs)}] Fetching {ref}") patients.append(fetch_patient_by_ref(access_token, ref)) return patients # ========================= # MODE B: Attempt system-wide Patient search paging # GET /Patient?_count=... # Follow Bundle.link[relation="next"] # ========================= def fetch_all_patients_by_search( access_token: str, count: int = 100, max_patients: int | None = None, elements: str | None = None, ) -> list[dict]: """ Tries GET /Patient with paging. Many Epic tenants will block/cap this. """ url = f"{FHIR_BASE}/Patient" params = {"_count": count} if elements: params["_elements"] = elements # optional, reduces payload if supported patients: list[dict] = [] while True: r = requests.get(url, headers=headers(access_token), params=params, timeout=30) # after first request, params must not be reused if url is already a "next" URL params = None # don't throw immediately; show OperationOutcome if present try: data = r.json() except Exception: r.raise_for_status() raise RuntimeError("Non-JSON response received") if not r.ok: print("HTTP:", r.status_code) print(json.dumps(data, indent=2)) raise RuntimeError("Search failed (see output above)") if is_operation_outcome(data): print(json.dumps(data, indent=2)) raise RuntimeError("OperationOutcome returned instead of Bundle") # Expect Bundle of Patients for entry in bundle_entries(data): res = entry.get("resource") if isinstance(res, dict) and res.get("resourceType") == "Patient": patients.append(res) if max_patients and len(patients) >= max_patients: return patients next_url = get_next_link(data) if not next_url: break url = next_url return patients # ========================= # CLI # ========================= def main(): parser = argparse.ArgumentParser(description="Fetch all Patients (Epic FHIR R4) via List or (if allowed) /Patient paging.") sub = parser.add_subparsers(dest="mode", required=True) # List mode (recommended) p_list = sub.add_parser("list", help="Fetch all patients from a List resource (recommended for Epic sandbox).") p_list.add_argument("--code", default="patients", help="List code (default: patients)") p_list.add_argument("--identifier", required=True, help="List identifier (e.g. urn:oid:...|5332)") p_list.add_argument("--out", default="patients.json", help="Output file (json or ndjson)") p_list.add_argument("--ndjson", action="store_true", help="Write NDJSON (one Patient per line)") # Search mode (only if tenant allows) p_search = sub.add_parser("search", help="Attempt GET /Patient with paging (often blocked/capped in Epic).") p_search.add_argument("--count", type=int, default=100, help="FHIR _count page size (default: 100)") p_search.add_argument("--max", type=int, default=None, help="Stop after N patients") p_search.add_argument("--elements", default=None, help="Optional _elements to reduce payload (e.g. id,name,gender,birthDate)") p_search.add_argument("--out", default="patients.json", help="Output file (json or ndjson)") p_search.add_argument("--ndjson", action="store_true", help="Write NDJSON (one Patient per line)") args = parser.parse_args() tok = get_token() access_token = tok["access_token"] if args.mode == "list": patients = fetch_all_patients_from_list(access_token, args.code, args.identifier) write_patients(args.out, patients, args.ndjson) print(f"\nSaved {len(patients)} patients to {args.out}") elif args.mode == "search": patients = fetch_all_patients_by_search( access_token, count=args.count, max_patients=args.max, elements=args.elements, ) write_patients(args.out, patients, args.ndjson) print(f"\nSaved {len(patients)} patients to {args.out}") if __name__ == "__main__": main()