Referral FHIR Server Documentation

This documentation provides detailed instructions for deploying, configuring, and customizing the Referral FHIR Server for Australian healthcare integration.

Table of Contents

• Overview
• Prerequisites
• Architecture
• AWS Deployment
  • Setting up API Gateway
  • Configuring Lambda Functions
  • PostgreSQL Integration
  • mTLS Configuration
  • Rate Limiting and Quotas
  • CORS Configuration
  • SQS Integration
• Security Considerations
• Python Code Customization
• Testing and Validation
• Alternative Deployment Options
• Troubleshooting

Overview

The Referral FHIR Server is a Python-based implementation designed to handle FHIR resources according to Australian healthcare profiles. It leverages AWS serverless technologies to provide a scalable, secure, and maintainable infrastructure for healthcare data exchange.

Prerequisites

Before you begin deployment, ensure you have:

• An AWS account with appropriate permissions
• AWS CLI installed and configured
• Python 3.9 or higher
• PostgreSQL database instance (RDS or externally hosted)
• Basic knowledge of FHIR standards and Australian healthcare profiles
• SSL certificates for mTLS configuration

Architecture

The Referral FHIR Server architecture consists of several key components:

1. API Gateway: Front-end service handling requests, authentication, and security
2. Lambda Functions: Python-based serverless functions that process FHIR payloads
3. Lambda Layers: Shared code for database connectivity and common utilities
4. PostgreSQL Database: Storage for FHIR resources
5. Optional SQS Queue: For guaranteed message delivery and processing

The flow of data through the system is as follows:

1. Client applications send FHIR-compliant requests to the API Gateway
2. API Gateway validates the request, performs authentication via mTLS
3. The request is routed to the appropriate Lambda function
4. The Lambda function processes the request, validates the FHIR payload
5. Data is stored or retrieved from PostgreSQL
6. If SQS is configured, messages are placed in a queue for guaranteed processing
7. Responses are returned to the client with appropriate status codes

AWS Deployment

Setting up API Gateway

1. Create a new API:

aws apigateway create-rest-api --name "ReferralFhirServer" --description "FHIR Server for Australian healthcare referrals"

2. Configure resources and methods:

Create resources for each FHIR endpoint and configure methods (GET, POST, PUT, DELETE) as needed. Example for creating a resource:

aws apigateway create-resource --rest-api-id <api-id> --parent-id <parent-resource-id> --path-part "Patient"

3. Set up request validation:

Create request validators to ensure proper formatting:

aws apigateway create-request-validator --rest-api-id <api-id> --name "FHIR-Validator" --validate-request-body --validate-request-parameters

Configuring Lambda Functions

1. Create Lambda Functions:

Create a Lambda function for each FHIR resource type you want to support:

aws lambda create-function \
  --function-name ReferralFhirPatient \
  --runtime python3.9 \
  --role arn:aws:iam::<account-id>:role/lambda-fhir-role \
  --handler lambda_function.handler \
  --zip-file fileb://function.zip

2. Create Lambda Layers:

Create Lambda layers for shared code, including the PostgreSQL connector:

aws lambda publish-layer-version \
  --layer-name pg-connector \
  --description "PostgreSQL connector for FHIR server" \
  --license-info "MIT" \
  --zip-file fileb://pg-connector-layer.zip \
  --compatible-runtimes python3.9 python3.10

3. Associate Lambda Layer with Functions:

aws lambda update-function-configuration \
  --function-name ReferralFhirPatient \
  --layers arn:aws:lambda:region:<account-id>:layer:pg-connector:<version>

PostgreSQL Integration

1. Create a Lambda Layer for PostgreSQL Connectivity:

Include the following in your PostgreSQL Lambda Layer:

• psycopg2-binary package for PostgreSQL connectivity
• Custom connection pool management
• Error handling and retry logic

2. Sample Database Connection Code:

import os
import psycopg2
from psycopg2.extras import RealDictCursor

def get_db_connection():
    try:
        conn = psycopg2.connect(
            host=os.environ['DB_HOST'],
            database=os.environ['DB_NAME'],
            user=os.environ['DB_USER'],
            password=os.environ['DB_PASSWORD'],
            cursor_factory=RealDictCursor
        )
        return conn
    except Exception as e:
        print(f"Error connecting to database: {e}")
        raise e

def execute_query(query, params=None):
    conn = get_db_connection()
    try:
        with conn.cursor() as cur:
            cur.execute(query, params)
            conn.commit()
            if cur.description:
                return cur.fetchall()
            return None
    finally:
        conn.close()

3. Store Database Credentials Securely:

Use AWS Secrets Manager to store database credentials:

aws secretsmanager create-secret \
  --name ReferralFhirDbCredentials \
  --description "Database credentials for FHIR Server" \
  --secret-string "{\"username\":\"dbuser\",\"password\":\"dbpassword\",\"engine\":\"postgres\",\"host\":\"dbinstance.example.region.rds.amazonaws.com\",\"port\":5432,\"dbname\":\"fhirdb\"}"

mTLS Configuration

1. Generate Certificates:

# Generate CA
openssl genrsa -out ca.key 2048
openssl req -new -x509 -days 365 -key ca.key -out ca.crt

# Generate Server Certificates
openssl genrsa -out server.key 2048
openssl req -new -key server.key -out server.csr
openssl x509 -req -days 365 -in server.csr -CA ca.crt -CAkey ca.key -set_serial 01 -out server.crt

# Generate Client Certificates
openssl genrsa -out client.key 2048
openssl req -new -key client.key -out client.csr
openssl x509 -req -days 365 -in client.csr -CA ca.crt -CAkey ca.key -set_serial 02 -out client.crt

2. Import Certificates to API Gateway:

aws apigateway import-client-certificate \
  --rest-api-id <api-id> \
  --body file://server.crt

3. Configure Custom Domain with mTLS:

aws apigateway create-domain-name \
  --domain-name api.fhir-referral.example.com \
  --regional-certificate-name <certificate-name> \
  --endpoint-configuration types=REGIONAL \
  --security-policy TLS_1_2 \
  --mutual-tls-authentication truststoreUri=s3://bucket-name/truststore.pem

4. Configure Base Path Mapping:

aws apigateway create-base-path-mapping \
  --domain-name api.fhir-referral.example.com \
  --rest-api-id <api-id> \
  --stage prod

Rate Limiting and Quotas

1. Configure Usage Plans and API Keys:

# Create an API key
aws apigateway create-api-key \
  --name "OrganizationAKey" \
  --description "API key for Organization A" \
  --enabled

# Create a usage plan
aws apigateway create-usage-plan \
  --name "StandardUsagePlan" \
  --description "Standard usage plan with rate limits" \
  --throttle burstLimit=20,rateLimit=10 \
  --quota limit=1000,offset=0,period=DAY

2. Associate API Key with Usage Plan:

aws apigateway create-usage-plan-key \
  --usage-plan-id <usage-plan-id> \
  --key-id <api-key-id> \
  --key-type "API_KEY"

CORS Configuration

1. Configure CORS on API Gateway:

aws apigateway put-integration-response \
  --rest-api-id <api-id> \
  --resource-id <resource-id> \
  --http-method GET \
  --status-code 200 \
  --response-parameters "{\"method.response.header.Access-Control-Allow-Origin\":\"'https://allowed-origin.example.com'\",\"method.response.header.Access-Control-Allow-Methods\":\"'GET,POST,PUT,DELETE'\",\"method.response.header.Access-Control-Allow-Headers\":\"'Content-Type,X-API-Key,Authorization'\"}" \
  --selection-pattern ""

2. Enable OPTIONS Method for CORS Pre-flight Requests:

aws apigateway put-method \
  --rest-api-id <api-id> \
  --resource-id <resource-id> \
  --http-method OPTIONS \
  --authorization-type "NONE"

SQS Integration

1. Create SQS Queue:

aws sqs create-queue --queue-name FhirReferralQueue

2. Modify Lambda to Use SQS:

import boto3
import json

sqs = boto3.client('sqs')
queue_url = 'https://sqs.region.amazonaws.com/account-id/FhirReferralQueue'

def lambda_handler(event, context):
    # Process the incoming FHIR request
    referral_data = json.loads(event['body'])
    
    # Validate the FHIR payload (implementation not shown)
    valid = validate_fhir_payload(referral_data)
    
    if not valid:
        return {
            'statusCode': 400,
            'body': json.dumps({'error': 'Invalid FHIR payload'})
        }
    
    # Send to SQS for guaranteed processing
    response = sqs.send_message(
        QueueUrl=queue_url,
        MessageBody=json.dumps(referral_data),
        MessageAttributes={
            'ReferralType': {
                'DataType': 'String',
                'StringValue': referral_data.get('resourceType', 'Unknown')
            }
        }
    )
    
    return {
        'statusCode': 202,
        'body': json.dumps({
            'message': 'Referral accepted for processing',
            'messageId': response['MessageId']
        })
    }

3. Create a Consumer Lambda Function:

def process_sqs_message(event, context):
    for record in event['Records']:
        payload = json.loads(record['body'])
        
        # Process the FHIR payload
        # Store in database, trigger workflows, etc.
        
        print(f"Processed message {record['messageId']}")

4. Configure Lambda Trigger from SQS:

aws lambda create-event-source-mapping \
  --function-name FhirReferralProcessor \
  --event-source-arn arn:aws:sqs:region:account-id:FhirReferralQueue

Security Considerations

Encryption

1. Data at Rest:

   • Enable encryption for PostgreSQL database
   • Use KMS for encrypting sensitive data in Lambda environment variables

2. Data in Transit:

   • Use HTTPS for all API communications
   • Configure mTLS for client authentication
   • Use minimum TLS 1.2 protocol

Access Control

1. IAM Policies:

   • Use least privilege principle for Lambda execution roles
   • Create separate roles for different functions based on their needs

2. Network Security:

   • Place PostgreSQL in a private subnet
   • Use Security Groups to restrict access
   • Consider using VPC endpoints for AWS services

Audit Logging

1. Enable CloudWatch Logs:

   • Configure detailed logging for API Gateway and Lambda
   • Set appropriate retention periods

2. Implement Audit Trails:

   • Log all FHIR operations with relevant metadata
   • Include user identifiers, timestamps, and request details

Example logging code:

import logging
import json
import uuid
import time

logger = logging.getLogger()
logger.setLevel(logging.INFO)

def audit_log(event_type, resource_type, resource_id, user_id, details=None):
    log_entry = {
        'timestamp': time.time(),
        'event_id': str(uuid.uuid4()),
        'event_type': event_type,
        'resource_type': resource_type,
        'resource_id': resource_id,
        'user_id': user_id,
        'details': details or {}
    }
    
    logger.info(json.dumps(log_entry))

Python Code Customization

Core Components

The FHIR server code consists of several Python modules:

1. lambda_function.py - Main Lambda handler
2. fhir_validator.py - FHIR payload validation
3. db_connector.py - Database operations
4. error_handler.py - Error processing and response formatting

Customizing FHIR Validation

The fhir_validator.py module can be extended to support specific Australian FHIR profiles:

import json
import os
import re
from jsonschema import validate, ValidationError

class FHIRValidator:
    def __init__(self):
        # Load schema definitions for Australian FHIR profiles
        self.schemas = {}
        schema_dir = os.path.join(os.path.dirname(__file__), 'schemas')
        for filename in os.listdir(schema_dir):
            if filename.endswith('.json'):
                resource_type = filename.split('.')[0]
                with open(os.path.join(schema_dir, filename)) as f:
                    self.schemas[resource_type] = json.load(f)
    
    def validate_resource(self, resource):
        """Validate a FHIR resource against Australian profiles"""
        if not isinstance(resource, dict):
            return False, "Resource must be a JSON object"
        
        resource_type = resource.get('resourceType')
        if not resource_type:
            return False, "Missing resourceType"
        
        if resource_type not in self.schemas:
            return False, f"Unsupported resourceType: {resource_type}"
        
        try:
            validate(instance=resource, schema=self.schemas[resource_type])
            return True, "Resource is valid"
        except ValidationError as e:
            return False, f"Validation error: {e.message}"
        
    def validate_australian_medicare_number(self, number):
        """Validate Australian Medicare number format"""
        if not number or not isinstance(number, str):
            return False
        
        # Medicare numbers are 10 or 11 digits
        pattern = r'^\d{10,11}$'
        return bool(re.match(pattern, number))
    
    def validate_australian_provider_number(self, number):
        """Validate Australian provider number format"""
        if not number or not isinstance(number, str):
            return False
        
        # Provider numbers are 8 characters: 6 digits followed by 2 letters or digits
        pattern = r'^\d{6}[0-9A-Z]{2}$'
        return bool(re.match(pattern, number))

Adding Custom FHIR Resources

To add support for additional FHIR resources:

1. Create a schema file in the schemas directory
2. Update the database schema to support the new resource
3. Create or modify Lambda functions to handle the resource

Example Lambda function for a custom resource:

import json
from fhir_validator import FHIRValidator
from db_connector import execute_query, get_db_connection
from error_handler import format_error

validator = FHIRValidator()

def handler(event, context):
    try:
        http_method = event['httpMethod']
        path_parameters = event.get('pathParameters', {}) or {}
        resource_id = path_parameters.get('id')
        
        if http_method == 'GET':
            if resource_id:
                return get_resource(resource_id)
            else:
                return search_resources(event.get('queryStringParameters', {}))
        elif http_method == 'POST':
            return create_resource(json.loads(event['body']))
        elif http_method == 'PUT':
            return update_resource(resource_id, json.loads(event['body']))
        elif http_method == 'DELETE':
            return delete_resource(resource_id)
        else:
            return {
                'statusCode': 405,
                'body': json.dumps({'error': 'Method not allowed'})
            }
    except Exception as e:
        return format_error(500, f"Internal server error: {str(e)}")

def get_resource(resource_id):
    result = execute_query(
        "SELECT data FROM fhir_resources WHERE id = %s AND resource_type = 'CustomResource'",
        (resource_id,)
    )
    
    if not result:
        return format_error(404, "Resource not found")
    
    return {
        'statusCode': 200,
        'body': json.dumps(result[0]['data'])
    }

# Implement other CRUD operations similarly

Testing and Validation

Local Testing

1. Set up a local environment:

# Create a Python virtual environment
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

# Install dependencies
pip install -r requirements.txt

# Run local tests
pytest

2. Test with AWS SAM Local:

# Install AWS SAM CLI
pip install aws-sam-cli

# Test a Lambda function locally
sam local invoke ReferralFhirPatient --event events/patient-create.json

Integration Testing

1. Deploy to a test environment:

# Create a test stage in API Gateway
aws apigateway create-deployment \
  --rest-api-id <api-id> \
  --stage-name test

2. Run integration tests:

import requests
import json
import unittest

class FHIRIntegrationTests(unittest.TestCase):
    BASE_URL = 'https://api.fhir-referral-test.example.com'
    API_KEY = 'your-api-key'
    
    def test_create_patient(self):
        # Load test patient data
        with open('test_data/patient.json') as f:
            patient_data = json.load(f)
        
        # Send request to API
        response = requests.post(
            f'{self.BASE_URL}/Patient',
            json=patient_data,
            headers={
                'Content-Type': 'application/json',
                'x-api-key': self.API_KEY
            },
            cert=('client.crt', 'client.key')
        )
        
        # Verify response
        self.assertEqual(response.status_code, 201)
        response_data = response.json()
        self.assertIn('id', response_data)
        
        # Clean up
        patient_id = response_data['id']
        delete_response = requests.delete(
            f'{self.BASE_URL}/Patient/{patient_id}',
            headers={'x-api-key': self.API_KEY},
            cert=('client.crt', 'client.key')
        )
        self.assertEqual(delete_response.status_code, 204)

if __name__ == '__main__':
    unittest.main()

Alternative Deployment Options

Vercel Deployment (Coming Soon)

1. Create a Vercel project:

   • Fork the GitHub repository
   • Connect to Vercel
   • Configure environment variables

2. Adapt the code for Vercel Serverless Functions:

// api/fhir/[resource].js
import { exec } from 'child_process';
import { promisify } from 'util';
const execAsync = promisify(exec);

export default async function handler(req, res) {
  try {
    const { resource } = req.query;
    const method = req.method;
    const body = req.body;
    
    // Call Python script with appropriate parameters
    const { stdout, stderr } = await execAsync(`python handler.py ${resource} ${method} '${JSON.stringify(body)}'`);
    
    if (stderr) {
      console.error(`Error: ${stderr}`);
      return res.status(500).json({ error: 'Internal server error' });
    }
    
    const result = JSON.parse(stdout);
    return res.status(result.statusCode).json(result.body);
  } catch (error) {
    console.error(error);
    return res.status(500).json({ error: 'Internal server error' });
  }
}

Azure Functions Deployment (Coming Soon)

1. Create Azure Function App:

az functionapp create \
  --resource-group myResourceGroup \
  --consumption-plan-location australiaeast \
  --runtime python \
  --runtime-version 3.9 \
  --functions-version 4 \
  --name fhir-referral-server \
  --storage-account fhirstorageaccount

2. Adapt the code for Azure Functions:

import azure.functions as func
import json
from fhir_validator import FHIRValidator
from db_connector import execute_query

app = func.FunctionApp()

@app.route(route="Patient/{id}")
def get_patient(req: func.HttpRequest) -> func.HttpResponse:
    patient_id = req.route_params.get('id')
    
    if not patient_id:
        return func.HttpResponse(
            json.dumps({"error": "Patient ID required"}),
            mimetype="application/json",
            status_code=400
        )
    
    result = execute_query(
        "SELECT data FROM fhir_resources WHERE id = %s AND resource_type = 'Patient'",
        (patient_id,)
    )
    
    if not result:
        return func.HttpResponse(
            json.dumps({"error": "Patient not found"}),
            mimetype="application/json",
            status_code=404
        )
    
    return func.HttpResponse(
        json.dumps(result[0]['data']),
        mimetype="application/json",
        status_code=200
    )

Troubleshooting

Common Issues and Solutions

1. API Gateway Authentication Errors:

   • Check mTLS certificate configuration
   • Verify API keys are properly set in request headers

2. Lambda Execution Errors:

   • Check CloudWatch logs for detailed error messages
   • Verify IAM permissions for Lambda functions

3. Database Connectivity Issues:

   • Check network security group settings
   • Verify database credentials are correctly stored in Secrets Manager

4. FHIR Validation Failures:

   • Enable detailed logging in the validator
   • Compare payloads against Australian FHIR profile specifications

Logging and Monitoring

1. Set up CloudWatch dashboards:

aws cloudwatch put-dashboard \
  --dashboard-name "FHIRServerMonitoring" \
  --dashboard-body file://dashboard.json

2. Configure CloudWatch alarms:

aws cloudwatch put-metric-alarm \
  --alarm-name "APILatencyAlarm" \
  --alarm-description "Alarm when API latency exceeds threshold" \
  --metric-name "Latency" \
  --namespace "AWS/ApiGateway" \
  --dimensions "Name=ApiId,Value=<api-id>" "Name=Stage,Value=prod" \
  --period 300 \
  --evaluation-periods 1 \
  --threshold 1000 \
  --comparison-operator "GreaterThanThreshold" \
  --statistic "Average" \
  --alarm-actions "arn:aws:sns:region:account-id:alert-topic"

For additional support and resources, visit our GitHub repository or contact our team at support@ausbiz.com.au.