Skip to navigation
Overview

Quickstart Code Example

Generate a synthetic patient and download its record and documents — in Python

This page asks for one patient with a few facts of your choosing, waits for the job to finish, and saves the patient’s FHIR bundle and clinical documents. It uses only the requests library.

Create an API key with the synth:process scope on the emtelligent platform, then set it in the environment:

pip install requests
export SYNTH_API_KEY='...'

Create a session

Authentication is a bearer token in the Authorization header: an API key, or an OAuth 2.0 access token.

import os
import time
import requests
BASE = "https://synth.emtelligent.com"
session = requests.Session()
session.headers["Authorization"] = f"Bearer {os.environ['SYNTH_API_KEY']}"

Submit a job

Every field of a request is optional: an empty request generates a random patient. This one describes the patient and asks for four documents, as a FHIR bundle and as text. The Idempotency-Key makes the submission safe to retry: sending the same key and body again returns the original job instead of creating, and charging for, a second one.

response = session.post(
    f"{BASE}/v1/jobs",
    headers={"Idempotency-Key": "quickstart-0001"},
    json={
        "patient": {
            "age": 64,
            "sex": "female",
            "conditions": [{"name": "type 2 diabetes mellitus"},
                           {"name": "chronic kidney disease stage 3"}],
            "medications": [{"name": "metformin 500 mg", "frequency": "twice daily"}],
            "smoking": {"status": "former", "start_age": 18, "quit_age": 45},
        },
        "complexity": 5,
        "num_documents": 4,
        "outputs": ["fhir", "notes"],
    },
)
response.raise_for_status()
job_id = response.json()["id"]

Wait for it to finish

A patient usually takes one to five minutes, depending on how many documents you ask for.

while True:
    job = session.get(f"{BASE}/v1/jobs/{job_id}").json()
    if job["status"] not in ("queued", "running"):
        break
    time.sleep(10)
if job["status"] != "succeeded":
    raise RuntimeError(f"job {job_id} {job['status']}: {job['error']}")
print(job["summary"]["counts"])
{'conditions': 5, 'medications': 5, 'allergies': 0, 'procedures': 0, 'immunizations': 6,
 'encounters': 4, 'observations': 48, 'notes': 4}

Download the results

Each result is listed in job["results"] with the URL to download it from.

for result in job["results"]:
    content = session.get(f"{BASE}{result['url']}").content
    with open(result["filename"], "wb") as f:
        f.write(content)
    print(f"saved {result['filename']} ({len(content):,} bytes)")
saved fhir_bundle.json (136,319 bytes)
saved notes.zip (5,848 bytes)
saved result.json (63,127 bytes)

Each result can be downloaded once: a second request returns 410 result_expired. Save it before you do anything else with it. Results not downloaded within 24 hours are deleted.

fhir_bundle.json is a FHIR R4 transaction bundle you can post to a FHIR server. notes.zip holds one text file per document, in date order, with a manifest.json describing each. result.json is the patient’s full record. See Results for the formats.

Next steps

Complete example

The steps above in one script. Save it as synth_quickstart.py, set SYNTH_API_KEY as at the top of this page, and run python synth_quickstart.py.

"""Generate a synthetic patient with the Synth API and save its results.
    pip install requests
    export SYNTH_API_KEY='...'
    python synth_quickstart.py
"""
import os
import sys
import time
import uuid
import requests
BASE = "https://synth.emtelligent.com"
session = requests.Session()
session.headers["Authorization"] = f"Bearer {os.environ['SYNTH_API_KEY']}"
REQUEST = {
    "patient": {
        "age": 64,
        "sex": "female",
        "conditions": [{"name": "type 2 diabetes mellitus"},
                       {"name": "chronic kidney disease stage 3"}],
        "medications": [{"name": "metformin 500 mg", "frequency": "twice daily"}],
        "smoking": {"status": "former", "start_age": 18, "quit_age": 45},
    },
    "complexity": 5,
    "num_documents": 4,
    "outputs": ["fhir", "notes"],
}
def call(method, path, **kwargs):
    """One API request. Waits and retries when the API asks it to (429, 503);
    exits with the API's error code for anything else that fails."""
    for attempt in range(5):
        response = session.request(method, f"{BASE}{path}", timeout=60, **kwargs)
        if response.status_code in (429, 503):
            time.sleep(int(response.headers.get("Retry-After", 2 ** attempt)))
            continue
        if response.status_code >= 400:
            try:
                error = response.json()["error"]
            except (ValueError, KeyError):
                error = {"code": "http_error", "message": response.text[:200]}
            sys.exit(f"{method} {path} failed: {response.status_code} {error.get('code')}: "
                     f"{error.get('message')} (request_id {error.get('request_id')})")
        return response
    sys.exit(f"{method} {path}: the API is still busy after 5 attempts")
def main():
    # The Idempotency-Key makes a retried submission return the same job instead of
    # creating (and charging for) a second one.
    job = call("POST", "/v1/jobs", headers={"Idempotency-Key": str(uuid.uuid4())}, json=REQUEST).json()
    print(f"submitted {job['id']}")
    # Wait for it to finish. A patient usually takes one to five minutes.
    deadline = time.monotonic() + 30 * 60
    while job["status"] in ("queued", "running"):
        if time.monotonic() > deadline:
            sys.exit(f"job {job['id']} is still {job['status']} after 30 minutes")
        time.sleep(10)
        job = call("GET", f"/v1/jobs/{job['id']}").json()
    if job["status"] != "succeeded":
        sys.exit(f"job {job['id']} {job['status']}: {job['error']}")
    print(f"generated patient {job['summary']['patient_id']}: {job['summary']['counts']}")
    # Download each result. This works once: save it before doing anything else.
    for result in job["results"]:
        content = call("GET", result["url"]).content
        filename = f"{job['id']}_{result['filename']}"
        with open(filename, "wb") as f:
            f.write(content)
        print(f"saved {filename} ({len(content):,} bytes)")
if __name__ == "__main__":
    main()

To generate many patients in one health system instead, see Generating a cohort.