Bulk FHIR API | Medplum

On this page

Medplum supports the Bulk FHIR API 2.0.0. The Bulk FHIR API uses Backend Services Authorization.

Use Cases

The premise of the Bulk FHIR API is that it allows you to create a bulk export of data for multiple patients. There are different ways to export data:

The export process is asynchronous, and you will need to poll a status URL returned when you start the export. After the BulkDataExport resource with the export results is available, it will contain a set of URLs where you can download the exported data in NDJSON format.

Access Policy Requirements

Because the bulk export process is asynchronous, your AccessPolicy must grant you access to the AsyncJob resourceType. This is required to poll the status of the export operation. Without access to AsyncJob, you will not be able to check the status of your export or retrieve the results.

Your AccessPolicy should include an entry like this:

{
  "resourceType": "AccessPolicy",
  "resource": [
    {
      "resourceType": "AsyncJob",
      "readonly": true
    }
  ]
}

Group Export

To specify which patients need to be included in the export, construct a Group resource and add specific patients as Group.member.entity. To start the process of exporting the resources, make an HTTP GET request for /fhir/R4/Group/<GROUP_ID>/$export?_outputFormat=ndjson. This initiates a Bulk Data Export transaction and return links to download URLs for requested resources.

curl 'https://api.medplum.com/fhir/R4/Group/<GROUP_ID>/$export?_outputFormat=ndjson' \
  -H 'Authorization: Bearer <ACCESS_TOKEN>'
Resource in Medplum App Usage in Bulk FHIR
Group All patients you want to include must be included as Group.member.entity

System Level Export

An export can also be performed for all resources in a Project by making a GET request for /fhir/R4/$export.

import http.client
import time
import json
import os
from typing import Any, TypedDict, List

class ExportOutput(TypedDict):
  type: str
  url: str

class BulkExportResponse(TypedDict):
  transactionTime: str
  request: str
  requiresAccessToken: bool
  output: List[ExportOutput]
  error: List[Any]

access_token = '[Requires valid access token]'
conn = http.client.HTTPSConnection('api.medplum.com')
conn.request(
  'GET', '/fhir/R4/$export', None, {
    'Authorization': 'Bearer ' + access_token,
    'Content-Type': 'application/fhir+json',
  })
init = conn.getresponse()
if init.status != 202:
  raise RuntimeError('Failed to start bulk export')
status_url: str | None = init.getheader('Content-Location')
if status_url == None:
  raise RuntimeError('No status URL found')
init.read()
conn.request(
  'GET', status_url, None, {
    'Authorization': 'Bearer ' + access_token,
  })
status = conn.getresponse()
while status.status == 202:
  status.read()
  time.sleep(1)
  conn.request(
    'GET', status_url, None, {
      'Authorization': 'Bearer ' + access_token,
    })
  status = conn.getresponse()
if status.status != 200:
  raise RuntimeError('Error exporting data')
body = status.read()
export: BulkExportResponse = json.loads(body)

def download_export_to_file(export_record: ExportOutput, access_token: str) -> None:
  from urllib.parse import urlparse
  url: str = export_record['url']
  parsed = urlparse(url)
  host = parsed.netloc
  path = parsed.path
  if parsed.query:
    path += '?' + parsed.query
  if parsed.scheme == 'https':
    download_conn = http.client.HTTPSConnection(host)
  else:
    download_conn = http.client.HTTPConnection(host)
  download_conn.request(
    'GET', path, None, {
      'Authorization': 'Bearer ' + access_token,
    })
  export_data = download_conn.getresponse()
  data: bytes = export_data.read()
  download_conn.close()
  file_path: str = os.path.join('medplum_resources', export_record['type'] + '.ndjson')
  with open(file_path, 'w') as f:
    f.write(data.decode('utf-8'))

os.makedirs('medplum_resources', exist_ok=True)
for record in export['output']:
  download_export_to_file(record, access_token)

Related Reading