healthcare hl7fhirhealthcareinteroperabilitysmart-on-fhirclinicalapiresources

hl7-fhir

Build healthcare integrations with HL7 FHIR: resources, operations, SMART on FHIR apps, and clinical data exchange. Use for health IT interoperability.

Quality 100% Code blocks 8 Sections 7 Lines 274 RU translation yes

view source

Models: sonnetopusgpt-6gemini-3glm-5

← hipaa-compliance

HL7 FHIR

Building healthcare integrations with HL7 FHIR resources and SMART on FHIR apps.

Quick Start

import requests

class FHIRClient:
    def __init__(self, base_url: str, access_token: str = None):
        self.base_url = base_url.rstrip('/')
        self.headers = {
            'Content-Type': 'application/fhir+json',
            'Accept': 'application/fhir+json',
        }
        if access_token:
            self.headers['Authorization'] = f'Bearer {access_token}'

    def read(self, resource_type: str, id: str) -> dict:
        """Read a FHIR resource by ID."""
        resp = requests.get(
            f'{self.base_url}/{resource_type}/{id}',
            headers=self.headers
        )
        resp.raise_for_status()
        return resp.json()

    def search(self, resource_type: str, params: dict = None) -> dict:
        """Search FHIR resources."""
        resp = requests.get(
            f'{self.base_url}/{resource_type}',
            params=params or {},
            headers=self.headers
        )
        resp.raise_for_status()
        return resp.json()

    def create(self, resource: dict) -> dict:
        """Create a new FHIR resource."""
        resp = requests.post(
            f'{self.base_url}/{resource["resourceType"]}',
            json=resource,
            headers=self.headers
        )
        resp.raise_for_status()
        return resp.json()

# Example: fetch a Patient
client = FHIRClient('https://fhir.example.com/fhir')
patient = client.read('Patient', '12345')
print(f"Patient: {patient['name'][0]['family']}")

When to Use

Building EHR integrations (Epic, Cerner, etc.)
Creating SMART on FHIR apps for clinical workflows
Exchanging clinical data between systems
When you need healthcare data interoperability

Step-by-Step

1. Understanding FHIR Resources

Common resources you'll work with:

# Patient resource
patient = {
    "resourceType": "Patient",
    "id": "12345",
    "name": [{
        "family": "Smith",
        "given": ["John"]
    }],
    "gender": "male",
    "birthDate": "1990-01-15"
}

# Observation (lab result)
observation = {
    "resourceType": "Observation",
    "status": "final",
    "code": {
        "coding": [{
            "system": "http://loinc.org",
            "code": "2339-0",
            "display": "Glucose [Mass/volume] in Blood"
        }]
    },
    "subject": {"reference": "Patient/12345"},
    "valueQuantity": {
        "value": 95,
        "unit": "mg/dL"
    }
}

2. SMART on FHIR Auth

import secrets
from urllib.parse import urlencode

class SMARTAuth:
    def __init__(self, client_id: str, client_secret: str, fhir_url: str):
        self.client_id = client_id
        self.client_secret = client_secret
        self.fhir_url = fhir_url

    def get_auth_url(self, redirect_uri: str, scopes: list[str]) -> str:
        """Generate OAuth2 authorization URL."""
        params = {
            'response_type': 'code',
            'client_id': self.client_id,
            'redirect_uri': redirect_uri,
            'scope': ' '.join(scopes),
            'state': secrets.token_urlsafe(32),
            'aud': self.fhir_url,
        }
        return f'{self.fhir_url}/auth/authorize?{urlencode(params)}'

    def exchange_code(self, code: str, redirect_uri: str) -> dict:
        """Exchange authorization code for access token."""
        resp = requests.post(f'{self.fhir_url}/auth/token', data={
            'grant_type': 'authorization_code',
            'code': code,
            'redirect_uri': redirect_uri,
            'client_id': self.client_id,
            'client_secret': self.client_secret,
        })
        return resp.json()

3. Clinical Queries

def get_patient_labs(client: FHIRClient, patient_id: str) -> list[dict]:
    """Fetch all lab results for a patient."""
    bundle = client.search('Observation', {
        'subject': f'Patient/{patient_id}',
        'category': 'laboratory',
        '_sort': '-date',
    })
    return bundle.get('entry', [])

def get_patient_medications(client: FHIRClient, patient_id: str) -> list[dict]:
    """Fetch active medications for a patient."""
    bundle = client.search('MedicationRequest', {
        'subject': f'Patient/{patient_id}',
        'status': 'active',
    })
    return bundle.get('entry', [])

def get_patient_conditions(client: FHIRClient, patient_id: str) -> list[dict]:
    """Fetch active conditions (diagnoses) for a patient."""
    bundle = client.search('Condition', {
        'subject': f'Patient/{patient_id}',
        'clinical-status': 'active',
    })
    return bundle.get('entry', [])

4. FHIR Bulk Data

def kick_off_bulk_export(client: FHIRClient, output_format: str = 'application/fhir+ndjson'):
    """Initiate FHIR Bulk Data export."""
    resp = requests.get(
        f'{client.base_url}/$export',
        headers={
            **client.headers,
            'Accept': output_format,
            'Prefer': 'respond-async',
        }
    )
    return resp.headers.get('Content-Location')  # Poll this URL for status

def check_bulk_status(status_url: str) -> dict:
    """Check status of bulk data export."""
    resp = requests.get(status_url)
    if resp.status_code == 200:
        return resp.json()  # Contains download URLs
    return {'status': 'in-progress'}

5. Subscriptions

def create_subscription(client: FHIRClient, criteria: str, endpoint: str) -> dict:
    """Create a FHIR subscription for real-time notifications."""
    subscription = {
        "resourceType": "Subscription",
        "status": "requested",
        "criteria": criteria,
        "channel": {
            "type": "rest-hook",
            "endpoint": endpoint,
            "payload": "application/fhir+json",
        }
    }
    return client.create(subscription)

# Example: subscribe to new Observations for a patient
create_subscription(
    client,
    "Observation?subject=Patient/12345",
    "https://myapp.example.com/fhir-webhook"
)

Best Practices

**Use official FHIR schemas** for validation
**Handle pagination** in search results (Bundle links)
**Cache resources** when appropriate to reduce server load
**Use _summary and _elements** to reduce payload size
**Implement retry logic** for transient failures
**Validate tokens** on every request in production

Common Pitfalls

Not handling Bundle pagination (missing results)
Ignoring FHIR version differences (R4 vs R5)
Not validating resource references before use
Forgetting to handle timezone in date/time fields
Not implementing proper OAuth2 token refresh

Examples

Patient Summary

def build_patient_summary(client: FHIRClient, patient_id: str) -> dict:
    """Build a comprehensive patient summary."""
    patient = client.read('Patient', patient_id)
    labs = get_patient_labs(client, patient_id)
    meds = get_patient_medications(client, patient_id)
    conditions = get_patient_conditions(client, patient_id)

    return {
        'patient': patient['name'][0],
        'recent_labs': [e['resource'] for e in labs[:10]],
        'active_medications': [e['resource'] for e in meds],
        'active_conditions': [e['resource'] for e in conditions],
    }

Validation

def test_fhir_client():
    client = FHIRClient('https://hapi.fhir.org/baseR4')
    patient = client.read('Patient', 'example')
    assert patient['resourceType'] == 'Patient'

def test_resource_validation():
    patient = {"resourceType": "Patient", "name": [{"family": "Test"}]}
    assert patient['resourceType'] == 'Patient'

← hipaa-compliance