Fhir Software

FHIR Software Development Skill

Expert guidance for building robust FHIR (Fast Healthcare Interoperability Resources) software systems with comprehensive package management, spec knowledge, and development workflows.

Core Architecture

1. Package/Specification Management

Local FHIR Package Cache:

• Use @fhir/package-loader or equivalent for TypeScript/Node.js environments
• For Python: fhir-package-loader or custom implementation using requests + json
• Cache strategy: ~/.fhir/packages/ with version-specific directories
• Support packages: hl7.fhir.r4.core, hl7.fhir.r5.core, Implementation Guides

Package Resolution Pattern:

// Load and cache FHIR packages
async function loadFhirPackage(packageId: string, version?: string) {
  const cacheDir = path.join(os.homedir(), '.fhir', 'packages', packageId, version || 'current');
  if (await fs.pathExists(cacheDir)) return loadFromCache(cacheDir);
  
  const packageData = await downloadPackage(packageId, version);
  await cachePackage(cacheDir, packageData);
  return packageData;
}

Document Index Structure: Build searchable index from package contents:

• StructureDefinitions (resources, profiles, extensions)
• SearchParameters (for API implementation)
• ValueSets and CodeSystems (terminology)
• OperationDefinitions (custom operations)
• CapabilityStatements (server capabilities)
• Example instances

2. Development Workflows

FHIR Resource Modeling

# Use Pydantic for FHIR resource modeling in Python
from pydantic import BaseModel, Field, field_validator
from typing import Optional, List, Literal
from enum import Enum
import re

class PatientGender(str, Enum):
    MALE = "male"
    FEMALE = "female"
    OTHER = "other"
    UNKNOWN = "unknown"

class Patient(BaseModel):
    resourceType: Literal["Patient"] = "Patient"
    id: Optional[str] = None
    active: Optional[bool] = None
    name: Optional[List[dict]] = None
    gender: Optional[PatientGender] = None
    birthDate: Optional[str] = None

    @field_validator('birthDate')
    @classmethod
    def validate_birthdate(cls, v):
        if v and not re.match(r'^\d{4}-\d{2}-\d{2}$', v):
            raise ValueError('Invalid date format, must be YYYY-MM-DD')
        return v

    class Config:
        extra = "allow"  # Allow additional FHIR elements

FHIR Server Implementation Patterns

FastAPI + Pydantic (Python):

from fastapi import FastAPI, HTTPException
from fhir.resources.patient import Patient

app = FastAPI()

@app.post("/Patient", response_model=Patient)
async def create_patient(patient: Patient):
    # Validate against FHIR spec
    patient.validate()
    # Store in database
    saved_patient = await db.save_patient(patient)
    return saved_patient

@app.get("/Patient/{patient_id}")
async def get_patient(patient_id: str):
    patient = await db.get_patient(patient_id)
    if not patient:
        raise HTTPException(404, "Patient not found")
    return patient

Express + FHIR TypeScript (Node.js):

import express from 'express';
import { Patient, Bundle } from 'fhir/r4';

const app = express();

app.post('/Patient', (req, res) => {
  const patient: Patient = req.body;
  
  // Validate resource type and required fields
  if (patient.resourceType !== 'Patient') {
    return res.status(400).json({
      resourceType: 'OperationOutcome',
      issue: [{
        severity: 'error',
        code: 'invalid',
        details: { text: 'Invalid resource type' }
      }]
    });
  }
  
  // Process and store
  const savedPatient = db.savePatient(patient);
  res.status(201).json(savedPatient);
});

3. FHIR Validation and Quality

Profile Validation

from fhir.resources.core.fhirabstractmodel import FHIRAbstractModel
from fhir.resources import get_fhir_model_class
import json

def validate_against_profile(resource_data: dict, profile_url: str) -> bool:
    """Validate FHIR resource against specific profile"""
    try:
        # Load profile from package cache
        profile = load_structure_definition(profile_url)

        # Validate using fhir.resources - dynamically get the resource class
        resource_type = resource_data.get('resourceType')
        resource_class = get_fhir_model_class(resource_type)
        resource = resource_class(**resource_data)

        # Additional profile-specific validation
        return validate_profile_constraints(resource, profile)
    except Exception as e:
        print(f"Validation error: {e}")
        return False

Terminology Validation

def validate_coding(coding: dict, value_set_url: str) -> bool:
    """Validate coding against ValueSet"""
    value_set = load_value_set(value_set_url)
    
    # Check if code exists in ValueSet expansion
    for concept in value_set.get('expansion', {}).get('contains', []):
        if (concept.get('code') == coding.get('code') and 
            concept.get('system') == coding.get('system')):
            return True
    return False

4. FHIR Search Implementation

Search Parameter Processing

from typing import Dict, Any
import re

class FhirSearchProcessor:
    def __init__(self):
        self.search_params = load_search_parameters()
    
    def parse_search_query(self, resource_type: str, params: Dict[str, str]) -> Dict[str, Any]:
        """Parse FHIR search parameters into database query"""
        query = {}
        
        for param_name, param_value in params.items():
            search_param = self.get_search_parameter(resource_type, param_name)
            
            if not search_param:
                continue
                
            # Handle different search parameter types
            if search_param['type'] == 'string':
                query[param_name] = self.parse_string_search(param_value)
            elif search_param['type'] == 'token':
                query[param_name] = self.parse_token_search(param_value)
            elif search_param['type'] == 'date':
                query[param_name] = self.parse_date_search(param_value)
            elif search_param['type'] == 'reference':
                query[param_name] = self.parse_reference_search(param_value)
                
        return query
    
    def parse_token_search(self, value: str) -> Dict[str, str]:
        """Parse token search: [system]|[code]"""
        if '|' in value:
            system, code = value.split('|', 1)
            result = {}
            if system:
                result['system'] = system
            if code:
                result['code'] = code
            return result
        return {'code': value} if value else {}

5. SMART on FHIR Integration

OAuth 2.0 / SMART App Launch

interface SmartConfig {
  authorizeUrl: string;
  tokenUrl: string;
  clientId: string;
  redirectUri: string;
  scopes: string[];
}

class SmartClient {
  constructor(private config: SmartConfig) {}
  
  async authorize(): Promise<string> {
    const state = generateRandomState();
    const authUrl = new URL(this.config.authorizeUrl);
    
    authUrl.searchParams.set('response_type', 'code');
    authUrl.searchParams.set('client_id', this.config.clientId);
    authUrl.searchParams.set('redirect_uri', this.config.redirectUri);
    authUrl.searchParams.set('scope', this.config.scopes.join(' '));
    authUrl.searchParams.set('state', state);
    authUrl.searchParams.set('aud', getFhirBaseUrl());
    
    return authUrl.toString();
  }
  
  async exchangeCodeForToken(code: string): Promise<TokenResponse> {
    const response = await fetch(this.config.tokenUrl, {
      method: 'POST',
      headers: { 'Content-Type': 'application/x-www-form-urlencoded' },
      body: new URLSearchParams({
        grant_type: 'authorization_code',
        client_id: this.config.clientId,
        code,
        redirect_uri: this.config.redirectUri
      })
    });

    if (!response.ok) {
      throw new Error(`Token exchange failed: ${response.statusText}`);
    }

    return response.json();
  }
}

Implementation Guides and Extensions

Custom Profile Development

{
  "resourceType": "StructureDefinition",
  "id": "my-patient-profile",
  "url": "http://example.org/fhir/StructureDefinition/MyPatient",
  "name": "MyPatientProfile",
  "status": "draft",
  "kind": "resource",
  "abstract": false,
  "type": "Patient",
  "baseDefinition": "http://hl7.org/fhir/StructureDefinition/Patient",
  "derivation": "constraint",
  "differential": {
    "element": [
      {
        "id": "Patient.identifier",
        "path": "Patient.identifier",
        "min": 1,
        "mustSupport": true
      },
      {
        "id": "Patient.name",
        "path": "Patient.name", 
        "min": 1,
        "max": "1"
      }
    ]
  }
}

Testing and Validation Tools

Unit Testing FHIR Resources

import pytest
from fhir.resources.patient import Patient

def test_patient_creation():
    patient_data = {
        "resourceType": "Patient",
        "id": "example",
        "active": True,
        "name": [{
            "family": "Doe",
            "given": ["John"]
        }],
        "gender": "male"
    }
    
    patient = Patient(**patient_data)
    assert patient.resourceType == "Patient"
    assert patient.gender == "male"
    assert len(patient.name) == 1
    assert patient.name[0].family == "Doe"

FHIR Server Testing

import httpx
import pytest

@pytest.fixture
async def fhir_client():
    async with httpx.AsyncClient(base_url="http://localhost:8000") as client:
        yield client

async def test_patient_crud(fhir_client):
    # Create patient
    patient_data = {"resourceType": "Patient", "active": True}
    create_response = await fhir_client.post("/Patient", json=patient_data)
    assert create_response.status_code == 201
    
    patient_id = create_response.json()["id"]
    
    # Read patient
    read_response = await fhir_client.get(f"/Patient/{patient_id}")
    assert read_response.status_code == 200
    
    # Update patient
    updated_data = read_response.json()
    updated_data["active"] = False
    update_response = await fhir_client.put(f"/Patient/{patient_id}", json=updated_data)
    assert update_response.status_code == 200
    
    # Delete patient
    delete_response = await fhir_client.delete(f"/Patient/{patient_id}")
    assert delete_response.status_code == 204

Common Patterns and Best Practices

Resource References and Includes

def resolve_references(resource: dict, include_params: List[str]) -> dict:
    """Resolve _include parameters for FHIR search"""
    included_resources = []
    
    for include_param in include_params:
        source_type, search_param = include_param.split(':', 1)
        
        if resource.get('resourceType') == source_type:
            ref_values = extract_reference_values(resource, search_param)
            
            for ref_value in ref_values:
                referenced_resource = load_resource_by_reference(ref_value)
                if referenced_resource:
                    included_resources.append(referenced_resource)
    
    return {
        'resourceType': 'Bundle',
        'type': 'searchset',
        'entry': [{'resource': resource}] + [{'resource': r} for r in included_resources]
    }

Batch and Transaction Processing

async def process_bundle(bundle: dict) -> dict:
    """Process FHIR Bundle with batch or transaction semantics"""
    response_entries = []
    transaction_mode = bundle.get('type') == 'transaction'
    
    try:
        if transaction_mode:
            await db.begin_transaction()
            
        for entry in bundle.get('entry', []):
            request = entry.get('request', {})
            resource = entry.get('resource')
            
            response_entry = await process_bundle_entry(request, resource)
            response_entries.append(response_entry)
            
        if transaction_mode:
            await db.commit_transaction()
            
    except Exception as e:
        if transaction_mode:
            await db.rollback_transaction()
        raise
    
    return {
        'resourceType': 'Bundle',
        'type': 'transaction-response' if transaction_mode else 'batch-response',
        'entry': response_entries
    }

Error Handling and OperationOutcome

def create_operation_outcome(severity: str, code: str, details: str) -> dict:
    """Create FHIR OperationOutcome for error reporting"""
    return {
        'resourceType': 'OperationOutcome',
        'issue': [{
            'severity': severity,
            'code': code,
            'details': {'text': details}
        }]
    }

# Usage in API endpoints
try:
    result = validate_fhir_resource(resource_data)
except ValidationError as e:
    return create_operation_outcome('error', 'invalid', str(e)), 400

Quick Reference Commands

Package Management

# Install FHIR packages
npm install @types/fhir @fhir/package-loader
pip install fhir.resources fhir-package-loader

# Load core FHIR packages
fhir-package-loader install hl7.fhir.r4.core 4.0.1
fhir-package-loader install hl7.fhir.us.core 5.0.1

Validation Tools

# FHIR Validator (Java)
java -jar validator_cli.jar resource.json -version 4.0.1

# HAPI FHIR Validator
curl -X POST "http://localhost:8080/fhir/$validate" \
  -H "Content-Type: application/fhir+json" \
  -d @patient.json

Development Server Setup

# Python FastAPI FHIR Server
uvicorn main:app --reload --port 8000

# Node.js Express FHIR Server  
npm start

Integration Points

• EHR Systems: Epic, Cerner, AllScripts FHIR APIs
• Cloud Platforms: AWS HealthLake, Azure FHIR, Google Healthcare API
• Terminology Services: UMLS, SNOMED CT, LOINC
• Security: OAuth 2.0, JWT, SMART on FHIR scopes
• Interoperability: HL7 v2 to FHIR conversion, CDA to FHIR

For detailed implementation guidance, reference the FHIR specification at https://hl7.org/fhir/ and implementation guides at https://fhir.org/guides/registry/