Medplum

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]


# You'll need to go through the auth process to get a valid access token,
# see https://www.medplum.com/docs/auth/client-credentials for details
access_token = '[Requires valid access token]'

# Open the connection to the Medplum API
conn = http.client.HTTPSConnection('api.medplum.com')

# Start the bulk export by calling the POST [base]/$export operation endpoint
# This begins the export process, which runs asynchronouosly and may take a while
# to finish.  Because of this, the response to this API call does not contain the
# actual exported data, but instead a URL that you can poll to get the status of
# the export operation
conn.request(
  'GET', '/fhir/R4/$export', None, {
    'Authorization': 'Bearer ' + access_token,
    'Content-Type': 'application/fhir+json',
  })
init = conn.getresponse()

# No 202 Accepted status code means the export request was not successfully started
if init.status != 202:
  raise RuntimeError('Failed to start bulk export')

# Get the status URL from the Content-Location header
status_url: str | None = init.getheader('Content-Location')
if status_url == None:
  raise RuntimeError('No status URL found')

# Read and discard the initial response body to allow reusing the connection
init.read()

# Make an initial request for the status of the export
conn.request(
  'GET', status_url, None, {
    'Authorization': 'Bearer ' + access_token,
  })
status = conn.getresponse()

# 202 Accepted status code means the export is still in progress
while status.status == 202:
  # Read and discard the response body before making the next request
  status.read()

  # Wait 1s between requests
  time.sleep(1)

  # Retry checking the status
  conn.request(
    'GET', status_url, None, {
      'Authorization': 'Bearer ' + access_token,
    })
  status = conn.getresponse()

# No 200 OK status code means the export failed with an error
if status.status != 200:
  raise RuntimeError('Error exporting data')

# Read the JSON body of the response
body = status.read()
export: BulkExportResponse = json.loads(body)

# The response JSON looks like this:
# {
#   "transactionTime": "2023-01-01T00:00:00Z",
#   "request" : "https://app.medplum.com/fhir/R4/$export
#   "requiresAccessToken" : true,
#   "output" : [{
#     "type" : "Patient",
#     "url" : "http://url.to.storage/patient_file_1.ndjson"
#   },{
#     "type" : "Observation",
#     "url" : "http://url.to.storage/observation_file_1.ndjson"
#   }],
#   "error" : []
# }
def download_export_to_file(export_record: ExportOutput, access_token: str) -> None:
  from urllib.parse import urlparse

  # Parse the URL to extract the host and path
  url: str = export_record['url']

  parsed = urlparse(url)
  host = parsed.netloc
  path = parsed.path
  if parsed.query:
    path += '?' + parsed.query

  # Create a new connection to the host specified in the URL
  if parsed.scheme == 'https':
    download_conn = http.client.HTTPSConnection(host)
  else:
    download_conn = http.client.HTTPConnection(host)

  # Request the NDJSON export data
  download_conn.request(
    'GET', path, None, {
      'Authorization': 'Bearer ' + access_token,
    })
  export_data = download_conn.getresponse()

  # Read the response once
  data: bytes = export_data.read()

  # Close the download connection
  download_conn.close()

  # Append NDJSON data to file on disk in medplum_resources folder
  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'))

# Create medplum_resources folder if it doesn't exist
os.makedirs('medplum_resources', exist_ok=True)

# Iterate over the output items to download the exported data
for record in export['output']:
  # record.type: the resource type contained in the export file
  # record.url: a URL pointing to an NDJSON file containing the exported data
  download_export_to_file(record, access_token)