docs(ehr): add EHR developer API guide and SDK examples (revived from fm/pix-4408)

developers/sdk-examples/python/auth.py

@@ -0,0 +1,57 @@
+"""Python SDK Example: Authentication
+
+Shows how to initialize the Pixelated client with API key auth,
+configure the base URL, and verify connectivity.
+
+Install: pip install pixelated_sdk
+"""
+
+import os
+
+from pixelated_sdk import ApiClient
+
+# ---------------------------------------------------------------------------
+# Basic client initialization
+# ---------------------------------------------------------------------------
+
+# The client sends the API key in the X-API-Key header on every request.
+# Set the key via an environment variable. Never hardcode it in source.
+
+client = ApiClient(
+    api_key=os.environ["PIXELATED_API_KEY"],
+    base_url="https://api.pixelated.com",
+)
+
+# ---------------------------------------------------------------------------
+# Sandbox client
+# ---------------------------------------------------------------------------
+
+# Use the sandbox for development. The base URL is the same, but the
+# API key routes requests to the sandbox tenant.
+
+sandbox_client = ApiClient(
+    api_key=os.environ["PIXELATED_SANDBOX_KEY"],
+    base_url="https://api.pixelated.com",
+)
+
+# ---------------------------------------------------------------------------
+# Verify connectivity
+# ---------------------------------------------------------------------------
+
+# If the key is invalid, the SDK raises an ApiException with status 401
+# and code "unauthorized". If the key lacks scope, you get 403 "forbidden".
+
+
+def verify_connection() -> None:
+    """Call a lightweight endpoint to confirm the key works."""
+    try:
+        patients = client.patients.list(limit=1)
+        total = patients.pagination.total
+        print(f"Connection OK. Patient count: {total}")
+    except Exception as err:
+        print(f"Connection failed: {err}")
+        raise
+
+
+if __name__ == "__main__":
+    verify_connection()

developers/sdk-examples/python/observations.py

@@ -0,0 +1,165 @@
+"""Python SDK Example: Observations
+
+Read, search, and create clinical observations through the EHR API.
+Observations include outcome measure scores like PHQ-9 and GAD-7.
+"""
+
+import os
+from datetime import datetime, timezone
+
+from pixelated_sdk import ApiClient
+
+client = ApiClient(
+    api_key=os.environ["PIXELATED_API_KEY"],
+    base_url="https://api.pixelated.com",
+)
+
+
+# ---------------------------------------------------------------------------
+# Search observations by patient
+# ---------------------------------------------------------------------------
+
+# At least one filter is required: patient, encounter, code, or status.
+# Combine patient with start/end to filter by date range.
+
+
+def search_by_patient(patient_id: str) -> None:
+    result = client.observations.list(patient=patient_id, limit=50, offset=0)
+
+    print(f"Found {result.pagination.total} observations")
+    for obs in result.data:
+        display = "Unknown"
+        if obs.code and obs.code.coding:
+            display = obs.code.coding[0].display or "Unknown"
+        value = obs.valueQuantity.value if obs.valueQuantity else "N/A"
+        print(f"  {obs.id}: {display} = {value}")
+
+
+# ---------------------------------------------------------------------------
+# Search observations by LOINC code
+# ---------------------------------------------------------------------------
+
+# Use the `code` parameter to filter by LOINC code. For example,
+# `44261-6` is the PHQ-9 total score.
+
+
+def search_by_code(patient_id: str, loinc_code: str) -> None:
+    result = client.observations.list(
+        patient=patient_id, code=loinc_code, limit=20
+    )
+
+    print(f"Found {len(result.data)} observations for code {loinc_code}")
+    for obs in result.data:
+        value = obs.valueQuantity.value if obs.valueQuantity else "N/A"
+        unit = obs.valueQuantity.unit if obs.valueQuantity else ""
+        print(f"  {obs.id}: value={value} {unit}")
+
+
+# ---------------------------------------------------------------------------
+# Search observations by date range
+# ---------------------------------------------------------------------------
+
+# Pass `patient` along with `start` and `end` (ISO timestamps) to
+# filter observations within a specific time window.
+
+
+def search_by_date_range(
+    patient_id: str, start_date: str, end_date: str
+) -> None:
+    result = client.observations.list(
+        patient=patient_id, start=start_date, end=end_date, limit=100
+    )
+
+    print(
+        f"Found {len(result.data)} observations "
+        f"between {start_date} and {end_date}"
+    )
+
+
+# ---------------------------------------------------------------------------
+# Read a single observation by ID
+# ---------------------------------------------------------------------------
+
+
+def read_observation(observation_id: str) -> None:
+    try:
+        obs = client.observations.read(observation_id)
+        print(f"Observation: {obs.data.id}, status={obs.data.status}")
+    except Exception as err:
+        if "not_found" in str(err):
+            print(f"Observation {observation_id} not found")
+            return
+        raise
+
+
+# ---------------------------------------------------------------------------
+# Create a PHQ-9 observation
+# ---------------------------------------------------------------------------
+
+# Send a FHIR R4 Observation resource. The example below records a
+# PHQ-9 total score of 12 (moderate depression).
+
+
+def create_phq9_observation(patient_id: str) -> None:
+    created = client.observations.create(
+        resourceType="Observation",
+        status="final",
+        subject={"reference": f"Patient/{patient_id}"},
+        code={
+            "coding": [
+                {
+                    "system": "http://loinc.org",
+                    "code": "44261-6",
+                    "display": "PHQ-9 total score",
+                }
+            ]
+        },
+        valueQuantity={
+            "value": 12,
+            "unit": "score",
+            "system": "http://unitsofmeasure.org",
+            "code": "{score}",
+        },
+        effectiveDateTime=datetime.now(timezone.utc).isoformat(),
+    )
+
+    print(f"Created observation with ID: {created.data.id}")
+
+
+# ---------------------------------------------------------------------------
+# Create a GAD-7 observation
+# ---------------------------------------------------------------------------
+
+# LOINC code 70274-6 is the GAD-7 total score.
+
+
+def create_gad7_observation(patient_id: str) -> None:
+    created = client.observations.create(
+        resourceType="Observation",
+        status="final",
+        subject={"reference": f"Patient/{patient_id}"},
+        code={
+            "coding": [
+                {
+                    "system": "http://loinc.org",
+                    "code": "70274-6",
+                    "display": "GAD-7 total score",
+                }
+            ]
+        },
+        valueQuantity={
+            "value": 8,
+            "unit": "score",
+            "system": "http://unitsofmeasure.org",
+            "code": "{score}",
+        },
+        effectiveDateTime=datetime.now(timezone.utc).isoformat(),
+    )
+
+    print(f"Created GAD-7 observation: {created.data.id}")
+
+
+if __name__ == "__main__":
+    patient_id = "patient-001"
+    search_by_patient(patient_id)
+    create_phq9_observation(patient_id)

developers/sdk-examples/python/patients.py

@@ -0,0 +1,137 @@
+"""Python SDK Example: Patients
+
+Read, search, create, and update patients through the EHR API.
+"""
+
+import os
+
+from pixelated_sdk import ApiClient
+
+client = ApiClient(
+    api_key=os.environ["PIXELATED_API_KEY"],
+    base_url="https://api.pixelated.com",
+)
+
+
+# ---------------------------------------------------------------------------
+# Search patients by name
+# ---------------------------------------------------------------------------
+
+# The `q` parameter matches against patient family and given names.
+# Results are paginated. Use `limit` and `offset` to page through.
+
+
+def search_patients() -> None:
+    result = client.patients.list(q="Smith", active="true", limit=20, offset=0)
+
+    print(f"Found {result.pagination.total} patients")
+    for patient in result.data:
+        name = patient.name[0] if patient.name else None
+        given = " ".join(name.given) if name and name.given else ""
+        family = name.family if name else ""
+        print(f"  {patient.id}: {given} {family}")
+
+
+# ---------------------------------------------------------------------------
+# Read a single patient by ID
+# ---------------------------------------------------------------------------
+
+# Returns the full FHIR Patient resource. Raises a not_found error
+# if the patient does not exist.
+
+
+def read_patient(patient_id: str) -> None:
+    try:
+        patient = client.patients.read(patient_id)
+        print(f"Patient: {patient.data.id}, active={patient.data.active}")
+    except Exception as err:
+        if "not_found" in str(err):
+            print(f"Patient {patient_id} not found")
+            return
+        raise
+
+
+# ---------------------------------------------------------------------------
+# Create a new patient
+# ---------------------------------------------------------------------------
+
+# Send a FHIR R4 Patient resource in the body. The response includes
+# the created patient with a server-assigned ID.
+
+
+def create_patient() -> None:
+    new_patient = client.patients.create(
+        name=[{"family": "Doe", "given": ["John"]}],
+        birthDate="1990-06-15",
+        gender="male",
+        telecom=[
+            {"system": "phone", "value": "555-0199"},
+            {"system": "email", "value": "john.doe@example.com"},
+        ],
+        address=[
+            {
+                "line": ["456 Oak Ave"],
+                "city": "Portland",
+                "state": "OR",
+                "postalCode": "97201",
+            }
+        ],
+        active=True,
+    )
+
+    print(f"Created patient with ID: {new_patient.data.id}")
+
+
+# ---------------------------------------------------------------------------
+# Update an existing patient
+# ---------------------------------------------------------------------------
+
+# Send the full patient resource. Partial updates are not supported.
+# The response returns the updated resource.
+
+
+def update_patient(patient_id: str) -> None:
+    updated = client.patients.update(
+        patient_id,
+        name=[{"family": "Doe", "given": ["John", "Michael"]}],
+        birthDate="1990-06-15",
+        gender="male",
+        active=True,
+    )
+
+    print(f"Updated patient: {updated.data.id}")
+
+
+# ---------------------------------------------------------------------------
+# Paginate through all active patients
+# ---------------------------------------------------------------------------
+
+# Loop until the data array is empty. Stop if `total` is available
+# and the offset exceeds it.
+
+
+def list_all_active_patients() -> None:
+    limit = 100
+    offset = 0
+    total = float("inf")
+    all_ids: list[str] = []
+
+    while offset < total:
+        page = client.patients.list(active="true", limit=limit, offset=offset)
+
+        for patient in page.data:
+            all_ids.append(patient.id)
+
+        if page.pagination.total is not None:
+            total = page.pagination.total
+
+        if len(page.data) < limit:
+            break
+        offset += limit
+
+    print(f"Total active patients: {len(all_ids)}")
+
+
+if __name__ == "__main__":
+    search_patients()
+    create_patient()