topologyhealth/claudefhirskill · GitHub
Comprehensive FHIR (Fast Healthcare Interoperability Resources) software development assistant. Use when working with FHIR APIs, implementations, or healthcare data exchange. Supports FHIR R4, R4B, R5, Implementation Guides (IGs), FHIR Shorthand (FSH) authoring, SUSHI, GoFSH, validation, terminology, and SMART on FHIR. Ideal for building FHIR servers, clients, validators, IG authors, or healthcare applications that need to process FHIR resources.
프로젝트 폴더에서 아래 명령어를 실행하고, 설치할 에이전트를 선택하세요.
npx skills add topologyhealth/claudefhirskill --skill fhir-software설치 명령을 직접 실행해야 적용됩니다. 지원 에이전트와 필요한 권한·라이선스는 제작자의 안내를 확인하세요.
.claude-plugin/marketplace.json{
"name": "fhir-developer-skills",
"owner": {
"name": "Topology Health",
"email": "info@topology.health"
},
"plugins": [
{
"name": "fhir-developer-skills",
"source": "./"
}
]
}
.claude-plugin/plugin.json{
"name": "fhir-developer-skills",
"version": "0.2.0",
"description": "Comprehensive FHIR software development skill covering FHIR R4/R5 APIs, resource modeling, server implementation, profile validation, terminology, SMART on FHIR, FSH authoring, SUSHI, GoFSH, and IG publishing",
"author": {
"name": "Topology Health",
"email": "info@topology.health"
},
"skills": [
"./"
]
}
.gitignore.DS_Store
LICENSE Apache License
Version 2.0, January 2004
http://www.apache.org/licenses/
TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
1. Definitions.
"License" shall mean the terms and conditions for use, reproduction,
and distribution as defined by Sections 1 through 9 of this document.
"Licensor" shall mean the copyright owner or entity authorized by
the copyright owner that is granting the License.
"Legal Entity" shall mean the union of the acting entity and all
other entities that control, are controlled by, or are under common
control with that entity. For the purposes of this definition,
"control" means (i) the power, direct or indirect, to cause the
direction or management of such entity, whether by contract or
otherwise, or (ii) ownership of fifty percent (50%) or more of the
outstanding shares, or (iii) beneficial ownership of such entity.
"You" (or "Your") shall mean an individual or Legal Entity
exercising permissions granted by this License.
"Source" form shall mean the preferred form for making modifications,
including but not limited to software source code, documentation
source, and configuration files.
"Object" form shall mean any form resulting from mechanical
transformation or translation of a Source form, including but
not limited to compiled object code, generated documentation,
and conversions to other media types.
"Work" shall mean the work of authorship, whether in Source or
Object form, made available under the License, as indicated by a
copyright notice that is included in or attached to the work
(an example is provided in the Appendix below).
"Derivative Works" shall mean any work, whether in Source or Object
form, that is based on (or derived from) the Work and for which the
editorial revisions, annotations, elaborations, or other modifications
represent, as a whole, an original work of authorship. For the purposes
of this License, Derivative Works shall not include works that remain
separable from, or merely link (or bind by name) to the interfaces of,
the Work and Derivative Works thereof.
"Contribution" shall mean any work of authorship, including
the original version of the Work and any modifications or additions
to that Work or Derivative Works thereof, that is intentionally
submitted to Licensor for inclusion in the Work by the copyright owner
or by an individual or Legal Entity authorized to submit on behalf of
the copyright owner. For the purposes of this definition, "submitted"
means any form of electronic, verbal, or written communication sent
to the Licensor or its representatives, including but not limited to
communication on electronic mailing lists, source code control systems,
and issue tracking systems that are managed by, or on behalf of, the
Licensor for the purpose of discussing and improving the Work, but
excluding communication that is conspicuously marked or otherwise
designated in writing by the copyright owner as "Not a Contribution."
"Contributor" shall mean Licensor and any individual or Legal Entity
on behalf of whom a Contribution has been received by Licensor and
subsequently incorporated within the Work.
2. Grant of Copyright License. Subject to the terms and conditions of
this License, each Contributor hereby grants to You a perpetual,
worldwide, non-exclusive, no-charge, royalty-free, irrevocable
copyright license to reproduce, prepare Derivative Works of,
publicly display, publicly perform, sublicense, and distribute the
Work and such Derivative Works in Source or Object form.
3. Grant of Patent License. Subject to the terms and conditions of
this License, each Contributor hereby grants to You a perpetual,
worldwide, non-exclusive, no-charge, royalty-free, irrevocable
(except as stated in this section) patent license to make, have made,
use, offer to sell, sell, import, and otherwise transfer the Work,
where such license applies only to those patent claims licensable
by such Contributor that are necessarily infringed by their
Contribution(s) alone or by combination of their Contribution(s)
with the Work to which such Contribution(s) was submitted. If You
institute patent litigation against any entity (including a
cross-claim or counterclaim in a lawsuit) alleging that the Work
or a Contribution incorporated within the Work constitutes direct
or contributory patent infringement, then any patent licenses
granted to You under this License for that Work shall terminate
as of the date such litigation is filed.
4. Redistribution. You may reproduce and distribute copies of the
Work or Derivative Works thereof in any medium, with or without
modifications, and in Source or Object form, provided that You
meet the following conditions:
(a) You must give any other recipients of the Work or
Derivative Works a copy of this License; and
(b) You must cause any modified files to carry prominent notices
stating that You changed the files; and
(c) You must retain, in the Source form of any Derivative Works
that You distribute, all copyright, patent, trademark, and
attribution notices from the Source form of the Work,
excluding those notices that do not pertain to any part of
the Derivative Works; and
(d) If the Work includes a "NOTICE" text file as part of its
distribution, then any Derivative Works that You distribute must
include a readable copy of the attribution notices contained
within such NOTICE file, excluding those notices that do not
pertain to any part of the Derivative Works, in at least one
of the following places: within a NOTICE text file distributed
as part of the Derivative Works; within the Source form or
documentation, if provided along with the Derivative Works; or,
within a display generated by the Derivative Works, if and
wherever such third-party notices normally appear. The contents
of the NOTICE file are for informational purposes only and
do not modify the License. You may add Your own attribution
notices within Derivative Works that You distribute, alongside
or as an addendum to the NOTICE text from the Work, provided
that such additional attribution notices cannot be construed
as modifying the License.
You may add Your own copyright statement to Your modifications and
may provide additional or different license terms and conditions
for use, reproduction, or distribution of Your modifications, or
for any such Derivative Works as a whole, provided Your use,
reproduction, and distribution of the Work otherwise complies with
the conditions stated in this License.
5. Submission of Contributions. Unless You explicitly state otherwise,
any Contribution intentionally submitted for inclusion in the Work
by You to the Licensor shall be under the terms and conditions of
this License, without any additional terms or conditions.
Notwithstanding the above, nothing herein shall supersede or modify
the terms of any separate license agreement you may have executed
with Licensor regarding such Contributions.
6. Trademarks. This License does not grant permission to use the trade
names, trademarks, service marks, or product names of the Licensor,
except as required for reasonable and customary use in describing the
origin of the Work and reproducing the content of the NOTICE file.
7. Disclaimer of Warranty. Unless required by applicable law or
agreed to in writing, Licensor provides the Work (and each
Contributor provides its Contributions) on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
implied, including, without limitation, any warranties or conditions
of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
PARTICULAR PURPOSE. You are solely responsible for determining the
appropriateness of using or redistributing the Work and assume any
risks associated with Your exercise of permissions under this License.
8. Limitation of Liability. In no event and under no legal theory,
whether in tort (including negligence), contract, or otherwise,
unless required by applicable law (such as deliberate and grossly
negligent acts) or agreed to in writing, shall any Contributor be
liable to You for damages, including any direct, indirect, special,
incidental, or consequential damages of any character arising as a
result of this License or out of the use or inability to use the
Work (including but not limited to damages for loss of goodwill,
work stoppage, computer failure or malfunction, or any and all
other commercial damages or losses), even if such Contributor
has been advised of the possibility of such damages.
9. Accepting Warranty or Additional Liability. While redistributing
the Work or Derivative Works thereof, You may choose to offer,
and charge a fee for, acceptance of support, warranty, indemnity,
or other liability obligations and/or rights consistent with this
License. However, in accepting such obligations, You may act only
on Your own behalf and on Your sole responsibility, not on behalf
of any other Contributor, and only if You agree to indemnify,
defend, and hold each Contributor harmless for any liability
incurred by, or claims asserted against, such Contributor by reason
of your accepting any such warranty or additional liability.
END OF TERMS AND CONDITIONS
APPENDIX: How to apply the Apache License to your work.
To apply the Apache License to your work, attach the following
boilerplate notice, with the fields enclosed by brackets "[]"
replaced with your own identifying information. (Don't include
the brackets!) The text should be enclosed in the appropriate
comment syntax for the file format. We also recommend that a
file or class name and description of purpose be included on the
same "printed page" as the copyright notice for easier
identification within third-party archives.
Copyright [yyyy] [name of copyright owner]
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
README.md# FHIR Software Development Skill A comprehensive Claude Code skill for building FHIR (Fast Healthcare Interoperability Resources) software systems with expert guidance on implementation, validation, and healthcare data exchange. ## Overview This skill provides specialized assistance for FHIR development across multiple versions (R4, R4B, R5) and programming languages. It includes expert knowledge of: - FHIR resource modeling and validation - Implementation Guide (IG) development - FHIR server and client implementation - SMART on FHIR integration - Terminology services and validation - Healthcare API development ## When to Use This Skill Invoke this skill when working on: - **FHIR API Development**: Building REST APIs that comply with FHIR specifications - **Healthcare Applications**: Apps that need to process or exchange FHIR resources - **FHIR Servers/Clients**: Implementing FHIR-compliant servers or client applications - **Data Validation**: Validating resources against FHIR profiles and Implementation Guides - **SMART on FHIR Apps**: Healthcare applications using SMART launch workflows - **Terminology Integration**: Working with ValueSets, CodeSystems, and terminology services - **Profile Development**: Creating custom FHIR profiles and Implementation Guides - **IG Authoring with FSH**: Writing FHIR Shorthand (`.fsh`) files, configuring SUSHI, and converting FHIR JSON to FSH with GoFSH ## Features ### Core Capabilities - **Package Management**: Guidance on using FHIR package loaders and managing local package caches - **Resource Modeling**: Best practices for modeling FHIR resources in TypeScript, Python, and other languages - **Server Implementation**: Patterns for FastAPI, Express, and other frameworks - **Search Implementation**: FHIR search parameter processing and query parsing - **Validation**: Profile validation, terminology validation, and constraint checking - **SMART on FHIR**: OAuth 2.0 workflows and app launch sequences - **Testing**: Unit testing patterns and integration testing for FHIR APIs ### Language Support - **TypeScript/Node.js**: Express, FHIR TypeScript types, package loaders - **Python**: FastAPI, Pydantic, fhir.resources library - Support for other languages with FHIR libraries ### FHIR Versions - FHIR R4 (4.0.1) - FHIR R4B - FHIR R5 - Implementation Guide compatibility ## Installation ### Prerequisites - [Claude Code](https://docs.anthropic.com/en/docs/claude-code) CLI or VS Code extension installed ### CLI 1. Launch Claude Code in your terminal: ```bash claude ``` 2. Add the marketplace: ```bash /plugin marketplace add TopologyHealth/ClaudeFHIRSkill ``` 3. Install the skill: ```bash /plugin install fhir-developer-skills@ClaudeFHIRSkill ``` ### VS Code 1. Click `/` in the Claude Code chat to open the command menu 2. Select **Manage Plugins** 3. Click **Marketplaces** 4. Enter the repository URL: `https://github.com/TopologyHealth/ClaudeFHIRSkill` 5. Click **Install** 6. The skill will appear in your Plugins list ### Manual Add this skill to your Claude Code environment by placing the `SKILL.md` file in your project's `.claude/skills/` directory or your global skills directory. ## Usage Examples ### Example 1: Building a FHIR Server Endpoint ```text I need to create a FHIR Patient endpoint in Python using FastAPI ``` The skill will provide guidance on: - Setting up FastAPI with fhir.resources - Implementing CRUD operations - Validation patterns - Error handling with OperationOutcome ### Example 2: Validating Against a Profile ```text How do I validate a Patient resource against the US Core Patient profile? ``` The skill will guide you through: - Loading the US Core package - Setting up profile validation - Checking must-support elements - Terminology binding validation ### Example 3: SMART on FHIR Integration ```text I need to implement SMART on FHIR authorization for my app ``` The skill provides: - OAuth 2.0 configuration - Authorization code flow implementation - Token exchange patterns - Scope management ## Key Patterns Covered - **Package/Specification Management**: Local caching, package resolution, document indexing - **Resource Modeling**: Type-safe models with validation - **Search Implementation**: Parameter parsing, query building, _include/_revinclude - **Batch/Transaction Processing**: Bundle handling with proper transaction semantics - **Error Handling**: Proper OperationOutcome generation - **Testing**: Unit and integration testing patterns ## Resources Referenced - [FHIR Specification](https://hl7.org/fhir/) - [Implementation Guide Registry](https://fhir.org/guides/registry/) - FHIR Package Registry - US Core Implementation Guide - SMART on FHIR Specification ## Contributing Contributions to improve this skill are welcome! Please ensure any additions: - Follow FHIR specification guidelines - Include working code examples - Cover common use cases - Support multiple programming languages where applicable ## License This skill is provided as-is for use with Claude Code. ## Support For issues or questions about using this skill with Claude Code, please refer to the [Claude Code documentation](https://code.claude.com/docs/). For FHIR specification questions, consult the [official FHIR documentation](https://hl7.org/fhir/) or the [FHIR community chat](https://chat.fhir.org/).
SKILL.md---
name: fhir-software
description: Comprehensive FHIR (Fast Healthcare Interoperability Resources) software development assistant. Use when working with FHIR APIs, implementations, or healthcare data exchange. Supports FHIR R4, R4B, R5, Implementation Guides (IGs), FHIR Shorthand (FSH) authoring, SUSHI, GoFSH, validation, terminology, and SMART on FHIR. Ideal for building FHIR servers, clients, validators, IG authors, or healthcare applications that need to process FHIR resources.
---
# 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:**
```typescript
// 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
```python
# 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):**
```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):**
```typescript
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
```python
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
```python
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
```python
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
```typescript
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
```json
{
"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
```python
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
```python
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
```python
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
```python
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
```python
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
```bash
# 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
```bash
# 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
```bash
# 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/
---
## 6. FHIR Shorthand (FSH) & IG Authoring
### Toolchain
| Tool | Role | Install |
|------|------|---------|
| **SUSHI** | Compiles `.fsh` files → FHIR JSON | `npm install -g fsh-sushi` |
| **GoFSH** | Converts FHIR JSON → FSH source | `npm install -g gofsh` |
| **IG Publisher** | Renders IG website from SUSHI output | Downloaded via `_updatePublisher.sh` |
### SUSHI Commands
```bash
sushi init # Scaffold new IG project
sushi build # Compile FSH → FHIR JSON (from project root)
sushi build . --snapshot # Include full StructureDefinition snapshot
sushi build . --log-level debug # Verbose output
sushi build . --preprocessed # Debug: dump resolved aliases/rulesets
sushi update-dependencies # Update deps to latest
```
Output: `fsh-generated/resources/{ResourceType}-{id}.json`. **SUSHI clears this folder on every build — never edit it manually.**
### GoFSH Commands
```bash
gofsh ./fhir-resources # Convert JSON artifacts to FSH
gofsh ./definitions -d hl7.fhir.us.core@6.1.0 # With extra dependencies
gofsh ./definitions --style group-by-profile # Organize output by profile
gofsh ./definitions --indent # Use indented rule style
gofsh ./definitions --fshing-trip # Validate round-trip accuracy
```
### Project Structure
```
my-ig/
├── sushi-config.yaml # Required: IG configuration
├── ig.ini # Required for IG Publisher
├── _genonce.sh / _genonce.bat # Run IG Publisher
├── _updatePublisher.sh / .bat # Download latest IG Publisher jar
├── input/
│ ├── fsh/ # All FSH source files
│ │ ├── aliases.fsh # Alias: $LNC = http://loinc.org
│ │ ├── profiles.fsh
│ │ ├── extensions.fsh
│ │ ├── valuesets.fsh
│ │ ├── codesystems.fsh
│ │ ├── instances.fsh
│ │ └── rulesets.fsh
│ ├── pagecontent/ # IG narrative (Markdown)
│ │ ├── index.md # Home page
│ │ ├── 1_background.md # Numbered = TOC order
│ │ ├── {resource-id}-intro.md # Content before artifact
│ │ └── {resource-id}-notes.md # Content after artifact
│ ├── images/ # Images, PDFs, spreadsheets
│ └── ignoreWarnings.txt
└── fsh-generated/ # SUSHI output (auto-generated, do not edit)
```
### sushi-config.yaml (Minimal Required Fields)
```yaml
id: hl7.fhir.us.example
canonical: http://hl7.org/fhir/us/example
name: ExampleIG
title: "Example Implementation Guide"
status: draft # draft | active | retired | unknown
version: 0.1.0
fhirVersion: 4.0.1
copyrightYear: 2024+
releaseLabel: ci-build
publisher:
name: My Organization
url: http://example.org
dependencies:
hl7.fhir.us.core: 6.1.0
menu:
Home: index.html
Artifacts: artifacts.html
parameters:
show-inherited-invariants: false
```
### FSH Entity Types
#### Profile
```
Profile: MyPatientProfile
Parent: Patient
Id: my-patient-profile
Title: "My Patient Profile"
Description: "Constrained Patient for our IG"
* identifier 1..* MS
* name 1..* MS
* birthDate MS
* gender 1..1 MS
* gender from http://hl7.org/fhir/ValueSet/administrative-gender (required)
```
#### Extension (Simple)
```
Extension: PatientReligion
Id: patient-religion
Title: "Patient Religion"
Context: Patient
* value[x] only CodeableConcept
* value[x] from ReligionValueSet (extensible)
```
#### Extension (Complex — sub-extensions)
```
Extension: USCoreEthnicityExtension
Id: us-core-ethnicity
Context: Patient, RelatedPerson, Practitioner
* extension contains
ombCategory 0..1 MS and
detailed 0..* and
text 1..1 MS
* extension[ombCategory].value[x] only Coding
* extension[ombCategory].value[x] from OmbEthnicityCategories (required)
* extension[text].value[x] only string
```
#### Instance
```
Instance: JaneDoe
InstanceOf: MyPatientProfile
Title: "Jane Doe"
Usage: #example // #example | #definition | #inline
* name[0].family = "Doe"
* name[0].given[0] = "Jane"
* birthDate = 1970-01-01
* gender = #female
```
#### ValueSet
```
ValueSet: MyConditionStatusVS
Id: my-condition-status
Title: "Condition Status Codes"
* include codes from system $SCT where concept is-a #404684003
* $V3#active "Active"
* exclude $SCT#74964007 "Other"
```
#### CodeSystem
```
CodeSystem: MyCustomCodes
Id: my-custom-codes
* #pending "Pending" "Awaiting review"
* #approved "Approved" "Formally approved"
// Hierarchical:
* #body "Body"
* #head "Head"
```
#### Invariant + Obeys
```
Invariant: my-inv-1
Description: "Value must be present for vital-signs category"
Severity: #error
Expression: "category.coding.code = 'vital-signs' implies value.exists()"
// In a Profile:
* obeys my-inv-1
```
#### RuleSet (Reusable / Parameterized)
```
RuleSet: PublicationMetadata
* ^status = #active
* ^experimental = false
* ^publisher = "My Org"
// Parameterized
RuleSet: SetContext(contextPath)
* ^context[+].type = #element
* ^context[=].expression = "{contextPath}"
// Usage:
* insert PublicationMetadata
* insert SetContext(Patient)
```
### FSH Rules Quick Reference
| Rule | Syntax | Example |
|------|--------|---------|
| Cardinality | `* elem min..max` | `* name 1..* MS` |
| Must Support | `* elem MS` | `* identifier MS` |
| Type constraint | `* elem only Type` | `* value[x] only Quantity` |
| Binding | `* elem from VS (strength)` | `* code from MyVS (required)` |
| Fixed value | `* elem = value` | `* status = #final` |
| Fixed coding | `* elem = $SYS#code "display"` | `* code = $LNC#29463-7` |
| Quantity | `* elem = n 'unit'` | `* value = 70 'kg'` |
| Slice (element) | `* arr contains name card` | `* component contains systolic 1..1` |
| Slice (extension) | `* extension contains Ext named n card` | `* extension contains $Race named race 0..1` |
| Obeys | `* obeys inv-id` | `* obeys us-core-6` |
| Caret (metadata) | `* ^property = val` | `* ^experimental = false` |
| Insert RuleSet | `* insert RSName` | `* insert PublicationMetadata` |
### Path Grammar Quick Reference
```
status // Simple element
name.family // Nested
valueQuantity // Choice [x] resolved
name[0] // Array index (zero-based)
name[+] // Soft index: next slot
name[=] // Soft index: same slot
component[respirationScore] // Slice by name
extension[race] // Extension slice
performer[Practitioner] // Reference target
^experimental // StructureDefinition metadata
code ^short // ElementDefinition property
```
### Coding / Quantity Syntax
```
// Alias declaration (top of file)
Alias: $LNC = http://loinc.org
Alias: $SCT = http://snomed.info/sct
// Code (no system)
#active
// Coding
http://loinc.org#29463-7 "Body Weight"
$LNC#29463-7 "Body Weight"
// Quantity (UCUM)
70.5 'kg' "kg"
120 'mm[Hg]' "mmHg"
```
### Naming Conventions
| Item | Convention | Example |
|------|-----------|---------|
| Profile/Extension/RuleSet names | `PascalCase` | `MyPatientProfile` |
| Item IDs | `kebab-case`, max 64 chars | `my-patient-profile` |
| Slice names | `lowerCamelCase` | `respirationScore` |
| Alias names | `$PrefixedName` | `$LNC`, `$SCT` |
### Key Rules
1. **Declare slices before constraining them** — `contains` rule must precede slice-specific rules
2. **Declare extensions before constraining sub-elements** — same ordering requirement
3. **`fsh-generated/` is owned by SUSHI** — never edit it; it is deleted and regenerated on each build
4. **Caret (`^`) rules are forbidden in Instances** — use them only in Profiles, Extensions, ValueSets, CodeSystems
### IG Authoring Workflow
```bash
# New IG from scratch
mkdir my-ig && cd my-ig
sushi init # Interactive scaffold
# Edit sushi-config.yaml and write .fsh files
sushi build # Compile
./_genonce.sh # Run IG Publisher
# Migrate existing FHIR JSON to FSH
gofsh ./existing-resources -d hl7.fhir.us.core@6.1.0 \
--style group-by-profile --indent --fshing-trip
# Debug a failing build
sushi build --log-level debug
sushi build --preprocessed # Inspect resolved aliases/rulesets
```
### Reference Files
- Full spec: https://build.fhir.org/ig/HL7/fhir-shorthand/reference.html
- SUSHI docs: https://fshschool.org/docs/sushi/
- GoFSH docs: https://fshschool.org/docs/gofsh/
- See also: `references/fsh_ig_authoring.md` for the complete reference
assets/bundle-searchset-template.json{
"resourceType": "Bundle",
"id": "search-results-example",
"meta": {
"lastUpdated": "2023-10-15T14:30:00Z"
},
"type": "searchset",
"total": 2,
"link": [
{
"relation": "self",
"url": "https://api.example.org/fhir/Patient?name=doe"
},
{
"relation": "first",
"url": "https://api.example.org/fhir/Patient?name=doe&_page=1"
},
{
"relation": "last",
"url": "https://api.example.org/fhir/Patient?name=doe&_page=1"
}
],
"entry": [
{
"fullUrl": "https://api.example.org/fhir/Patient/example-patient",
"resource": {
"resourceType": "Patient",
"id": "example-patient",
"meta": {
"versionId": "1",
"lastUpdated": "2023-10-15T12:00:00Z"
},
"identifier": [
{
"system": "http://hospital.example.org/patients",
"value": "123456789"
}
],
"active": true,
"name": [
{
"family": "Doe",
"given": ["John"]
}
],
"gender": "male",
"birthDate": "1980-01-01"
},
"search": {
"mode": "match",
"score": 1.0
}
},
{
"fullUrl": "https://api.example.org/fhir/Patient/example-patient-2",
"resource": {
"resourceType": "Patient",
"id": "example-patient-2",
"meta": {
"versionId": "1",
"lastUpdated": "2023-10-15T11:30:00Z"
},
"identifier": [
{
"system": "http://hospital.example.org/patients",
"value": "987654321"
}
],
"active": true,
"name": [
{
"family": "Doe",
"given": ["Jane"]
}
],
"gender": "female",
"birthDate": "1975-05-15"
},
"search": {
"mode": "match",
"score": 0.9
}
}
]
}
assets/capability-statement-template.json{
"resourceType": "CapabilityStatement",
"id": "example-fhir-server",
"meta": {
"lastUpdated": "2023-10-15T14:30:00Z"
},
"url": "https://api.example.org/fhir/metadata",
"version": "1.0.0",
"name": "ExampleFHIRServer",
"title": "Example FHIR Server Capability Statement",
"status": "active",
"experimental": false,
"date": "2023-10-15",
"publisher": "Example Healthcare Organization",
"contact": [
{
"name": "FHIR Support Team",
"telecom": [
{
"system": "email",
"value": "fhir-support@example.org"
}
]
}
],
"description": "Capability statement for the Example FHIR Server supporting patient management and clinical data exchange",
"kind": "instance",
"software": {
"name": "Example FHIR Server",
"version": "2.1.0",
"releaseDate": "2023-10-01"
},
"implementation": {
"description": "Example FHIR Server for healthcare data management",
"url": "https://api.example.org/fhir"
},
"fhirVersion": "4.0.1",
"format": [
"application/fhir+json",
"application/fhir+xml"
],
"patchFormat": [
"application/json-patch+json",
"application/fhir+json"
],
"rest": [
{
"mode": "server",
"documentation": "Main FHIR endpoint supporting patient and clinical resources",
"security": {
"cors": true,
"service": [
{
"coding": [
{
"system": "http://terminology.hl7.org/CodeSystem/restful-security-service",
"code": "OAuth",
"display": "OAuth2 using SMART-on-FHIR profile"
}
]
}
],
"description": "Uses OAuth2/SMART-on-FHIR for authorization"
},
"resource": [
{
"type": "Patient",
"profile": "http://hl7.org/fhir/us/core/StructureDefinition/us-core-patient",
"documentation": "Patient resource supporting US Core profile",
"interaction": [
{"code": "read"},
{"code": "vread"},
{"code": "update"},
{"code": "delete"},
{"code": "create"},
{"code": "search-type"}
],
"versioning": "versioned-update",
"readHistory": true,
"updateCreate": false,
"conditionalCreate": true,
"conditionalUpdate": true,
"conditionalDelete": "multiple",
"searchParam": [
{
"name": "identifier",
"type": "token",
"documentation": "A patient identifier"
},
{
"name": "name",
"type": "string",
"documentation": "Patient name"
},
{
"name": "birthdate",
"type": "date",
"documentation": "Date of birth"
},
{
"name": "gender",
"type": "token",
"documentation": "Gender"
}
]
},
{
"type": "Observation",
"profile": "http://hl7.org/fhir/us/core/StructureDefinition/us-core-observation-lab",
"interaction": [
{"code": "read"},
{"code": "create"},
{"code": "search-type"}
],
"searchParam": [
{
"name": "subject",
"type": "reference",
"documentation": "Subject of observation"
},
{
"name": "code",
"type": "token",
"documentation": "Observation code"
},
{
"name": "date",
"type": "date",
"documentation": "Observation date"
}
]
}
],
"interaction": [
{
"code": "transaction",
"documentation": "Support for FHIR transactions"
},
{
"code": "batch",
"documentation": "Support for FHIR batch operations"
}
]
}
]
}
assets/fhir_server.py#!/usr/bin/env python3
"""
Basic FHIR Server using FastAPI
Provides a foundation for building FHIR-compliant APIs
"""
from fastapi import FastAPI, HTTPException, Depends, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from typing import Dict, Any, Optional, List
import uvicorn
from datetime import datetime
import json
import os
# FHIR imports
from fhir.resources.patient import Patient
from fhir.resources.observation import Observation
from fhir.resources.bundle import Bundle, BundleEntry
from fhir.resources.capabilitystatement import CapabilityStatement
from fhir.resources.operationoutcome import OperationOutcome, OperationOutcomeIssue
from pydantic import ValidationError
app = FastAPI(
title="Example FHIR Server",
description="Basic FHIR R4 compliant server",
version="1.0.0",
docs_url="/docs"
)
# CORS middleware
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# In-memory storage (use a real database in production)
patients_db: Dict[str, Dict] = {}
observations_db: Dict[str, Dict] = {}
# Utility functions
def create_operation_outcome(severity: str, code: str, details: str) -> Dict:
"""Create FHIR OperationOutcome for error responses"""
issue = OperationOutcomeIssue(
severity=severity,
code=code,
details={"text": details}
)
outcome = OperationOutcome(issue=[issue])
return outcome.dict(exclude_none=True)
def generate_id() -> str:
"""Generate a simple ID for resources"""
from uuid import uuid4
return str(uuid4())
def validate_fhir_resource(resource_data: Dict, resource_class) -> Any:
"""Validate FHIR resource using Pydantic models"""
try:
return resource_class(**resource_data)
except ValidationError as e:
raise HTTPException(
status_code=400,
detail=create_operation_outcome("error", "invalid", str(e))
)
# Dependency for FHIR content type
async def validate_fhir_content_type(request: Request):
content_type = request.headers.get("content-type", "")
if request.method in ["POST", "PUT"] and "application/fhir+json" not in content_type:
if "application/json" not in content_type:
raise HTTPException(
status_code=415,
detail=create_operation_outcome(
"error", "not-supported",
"Content-Type must be application/fhir+json or application/json"
)
)
# Metadata endpoint
@app.get("/metadata", response_model=Dict)
async def get_capability_statement():
"""Return server capability statement"""
capability = {
"resourceType": "CapabilityStatement",
"status": "active",
"date": datetime.utcnow().isoformat() + "Z",
"publisher": "Example FHIR Server",
"kind": "instance",
"software": {
"name": "FastAPI FHIR Server",
"version": "1.0.0"
},
"fhirVersion": "4.0.1",
"format": ["application/fhir+json"],
"rest": [{
"mode": "server",
"resource": [
{
"type": "Patient",
"interaction": [
{"code": "read"},
{"code": "create"},
{"code": "update"},
{"code": "delete"},
{"code": "search-type"}
],
"searchParam": [
{"name": "name", "type": "string"},
{"name": "family", "type": "string"},
{"name": "given", "type": "string"},
{"name": "birthdate", "type": "date"},
{"name": "gender", "type": "token"}
]
},
{
"type": "Observation",
"interaction": [
{"code": "read"},
{"code": "create"},
{"code": "search-type"}
],
"searchParam": [
{"name": "subject", "type": "reference"},
{"name": "patient", "type": "reference"},
{"name": "code", "type": "token"},
{"name": "date", "type": "date"}
]
}
]
}]
}
return capability
# Patient endpoints
@app.post("/Patient", dependencies=[Depends(validate_fhir_content_type)])
async def create_patient(patient_data: Dict[str, Any]):
"""Create a new patient"""
patient = validate_fhir_resource(patient_data, Patient)
# Generate ID if not provided
if not patient.id:
patient.id = generate_id()
# Add metadata
patient.meta = {
"versionId": "1",
"lastUpdated": datetime.utcnow().isoformat() + "Z"
}
# Store in database
patients_db[patient.id] = patient.dict(exclude_none=True)
return JSONResponse(
status_code=201,
content=patients_db[patient.id],
headers={"Location": f"/Patient/{patient.id}"}
)
@app.get("/Patient/{patient_id}")
async def get_patient(patient_id: str):
"""Read a patient by ID"""
if patient_id not in patients_db:
raise HTTPException(
status_code=404,
detail=create_operation_outcome("error", "not-found", f"Patient/{patient_id} not found")
)
return patients_db[patient_id]
@app.put("/Patient/{patient_id}", dependencies=[Depends(validate_fhir_content_type)])
async def update_patient(patient_id: str, patient_data: Dict[str, Any]):
"""Update an existing patient"""
patient = validate_fhir_resource(patient_data, Patient)
# Ensure ID matches URL
patient.id = patient_id
# Update metadata
existing = patients_db.get(patient_id, {})
current_version = int(existing.get("meta", {}).get("versionId", "0"))
patient.meta = {
"versionId": str(current_version + 1),
"lastUpdated": datetime.utcnow().isoformat() + "Z"
}
# Store in database
patients_db[patient_id] = patient.dict(exclude_none=True)
return patients_db[patient_id]
@app.delete("/Patient/{patient_id}")
async def delete_patient(patient_id: str):
"""Delete a patient"""
if patient_id not in patients_db:
raise HTTPException(
status_code=404,
detail=create_operation_outcome("error", "not-found", f"Patient/{patient_id} not found")
)
del patients_db[patient_id]
return JSONResponse(status_code=204, content=None)
@app.get("/Patient")
async def search_patients(
name: Optional[str] = None,
family: Optional[str] = None,
given: Optional[str] = None,
birthdate: Optional[str] = None,
gender: Optional[str] = None,
_count: Optional[int] = 20,
_offset: Optional[int] = 0
):
"""Search for patients"""
results = []
for patient_data in patients_db.values():
match = True
# Simple string matching for name fields
if name and patient_data.get("name"):
name_match = any(
name.lower() in (n.get("family", "") + " " + " ".join(n.get("given", []))).lower()
for n in patient_data["name"]
)
if not name_match:
match = False
if family and patient_data.get("name"):
family_match = any(
family.lower() in n.get("family", "").lower()
for n in patient_data["name"]
)
if not family_match:
match = False
if given and patient_data.get("name"):
given_match = any(
any(given.lower() in g.lower() for g in n.get("given", []))
for n in patient_data["name"]
)
if not given_match:
match = False
if birthdate and patient_data.get("birthDate") != birthdate:
match = False
if gender and patient_data.get("gender") != gender:
match = False
if match:
results.append(patient_data)
# Apply pagination
total = len(results)
paginated_results = results[_offset:_offset + _count]
# Create Bundle
bundle = {
"resourceType": "Bundle",
"type": "searchset",
"total": total,
"entry": [
{
"resource": result,
"search": {"mode": "match"}
}
for result in paginated_results
]
}
return bundle
# Observation endpoints
@app.post("/Observation", dependencies=[Depends(validate_fhir_content_type)])
async def create_observation(observation_data: Dict[str, Any]):
"""Create a new observation"""
observation = validate_fhir_resource(observation_data, Observation)
if not observation.id:
observation.id = generate_id()
observation.meta = {
"versionId": "1",
"lastUpdated": datetime.utcnow().isoformat() + "Z"
}
observations_db[observation.id] = observation.dict(exclude_none=True)
return JSONResponse(
status_code=201,
content=observations_db[observation.id],
headers={"Location": f"/Observation/{observation.id}"}
)
@app.get("/Observation/{observation_id}")
async def get_observation(observation_id: str):
"""Read an observation by ID"""
if observation_id not in observations_db:
raise HTTPException(
status_code=404,
detail=create_operation_outcome("error", "not-found", f"Observation/{observation_id} not found")
)
return observations_db[observation_id]
@app.get("/Observation")
async def search_observations(
subject: Optional[str] = None,
patient: Optional[str] = None,
code: Optional[str] = None,
date: Optional[str] = None,
_count: Optional[int] = 20
):
"""Search for observations"""
results = []
for obs_data in observations_db.values():
match = True
# Subject/patient reference matching
if subject and obs_data.get("subject", {}).get("reference") != subject:
match = False
if patient and obs_data.get("subject", {}).get("reference") != f"Patient/{patient}":
match = False
# Simple code matching
if code and obs_data.get("code"):
code_match = any(
code in coding.get("code", "")
for coding in obs_data["code"].get("coding", [])
)
if not code_match:
match = False
# Date matching (simplified)
if date:
obs_date = obs_data.get("effectiveDateTime", "")
if not obs_date.startswith(date):
match = False
if match:
results.append(obs_data)
bundle = {
"resourceType": "Bundle",
"type": "searchset",
"total": len(results),
"entry": [
{
"resource": result,
"search": {"mode": "match"}
}
for result in results[:_count]
]
}
return bundle
# Health check
@app.get("/health")
async def health_check():
"""Health check endpoint"""
return {"status": "healthy", "timestamp": datetime.utcnow().isoformat()}
if __name__ == "__main__":
port = int(os.getenv("PORT", 8000))
uvicorn.run(app, host="0.0.0.0", port=port)
assets/observation-vitals-template.json{
"resourceType": "Observation",
"id": "example-vitals",
"meta": {
"profile": ["http://hl7.org/fhir/us/core/StructureDefinition/us-core-vital-signs"]
},
"status": "final",
"category": [
{
"coding": [
{
"system": "http://terminology.hl7.org/CodeSystem/observation-category",
"code": "vital-signs",
"display": "Vital Signs"
}
]
}
],
"code": {
"coding": [
{
"system": "http://loinc.org",
"code": "85354-9",
"display": "Blood pressure panel with all children optional"
}
]
},
"subject": {
"reference": "Patient/example-patient",
"display": "John Doe"
},
"encounter": {
"reference": "Encounter/example-encounter"
},
"effectiveDateTime": "2023-10-15T14:30:00Z",
"performer": [
{
"reference": "Practitioner/example-practitioner",
"display": "Dr. Smith"
}
],
"component": [
{
"code": {
"coding": [
{
"system": "http://loinc.org",
"code": "8480-6",
"display": "Systolic blood pressure"
}
]
},
"valueQuantity": {
"value": 120,
"unit": "mmHg",
"system": "http://unitsofmeasure.org",
"code": "mm[Hg]"
}
},
{
"code": {
"coding": [
{
"system": "http://loinc.org",
"code": "8462-4",
"display": "Diastolic blood pressure"
}
]
},
"valueQuantity": {
"value": 80,
"unit": "mmHg",
"system": "http://unitsofmeasure.org",
"code": "mm[Hg]"
}
}
]
}
assets/package.json{
"name": "fhir-project",
"version": "1.0.0",
"description": "FHIR implementation project",
"main": "dist/index.js",
"scripts": {
"start": "node dist/index.js",
"dev": "tsx src/index.ts",
"build": "tsc",
"test": "jest",
"lint": "eslint src/**/*.ts",
"validate": "node scripts/fhir_validator.js"
},
"keywords": ["fhir", "healthcare", "hl7"],
"author": "",
"license": "MIT",
"dependencies": {
"@types/fhir": "^0.0.37",
"fhir": "^4.11.1",
"@fhir/package-loader": "^1.0.0",
"express": "^4.18.2",
"cors": "^2.8.5",
"helmet": "^7.0.0",
"dotenv": "^16.3.1",
"jsonwebtoken": "^9.0.2",
"node-fetch": "^3.3.2"
},
"devDependencies": {
"@types/express": "^4.17.17",
"@types/cors": "^2.8.13",
"@types/jsonwebtoken": "^9.0.2",
"@types/node": "^20.5.0",
"typescript": "^5.1.6",
"tsx": "^3.12.7",
"jest": "^29.6.2",
"@types/jest": "^29.5.3",
"eslint": "^8.46.0",
"@typescript-eslint/eslint-plugin": "^6.2.1",
"@typescript-eslint/parser": "^6.2.1"
},
"engines": {
"node": ">=18.0.0"
}
}
assets/patient-template.json{
"resourceType": "Patient",
"id": "example-patient",
"meta": {
"profile": ["http://hl7.org/fhir/us/core/StructureDefinition/us-core-patient"]
},
"identifier": [
{
"use": "usual",
"type": {
"coding": [
{
"system": "http://terminology.hl7.org/CodeSystem/v2-0203",
"code": "MR",
"display": "Medical Record Number"
}
]
},
"system": "http://hospital.example.org/patients",
"value": "123456789"
}
],
"active": true,
"name": [
{
"use": "official",
"family": "Doe",
"given": ["John", "Michael"]
}
],
"telecom": [
{
"system": "phone",
"value": "+1-555-123-4567",
"use": "home"
},
{
"system": "email",
"value": "john.doe@example.com",
"use": "home"
}
],
"gender": "male",
"birthDate": "1980-01-01",
"address": [
{
"use": "home",
"type": "both",
"line": ["123 Main Street", "Apt 4B"],
"city": "Springfield",
"state": "IL",
"postalCode": "62701",
"country": "US"
}
],
"contact": [
{
"relationship": [
{
"coding": [
{
"system": "http://terminology.hl7.org/CodeSystem/v2-0131",
"code": "C",
"display": "Emergency Contact"
}
]
}
],
"name": {
"family": "Doe",
"given": ["Jane"]
},
"telecom": [
{
"system": "phone",
"value": "+1-555-987-6543"
}
]
}
],
"communication": [
{
"language": {
"coding": [
{
"system": "urn:ietf:bcp:47",
"code": "en-US",
"display": "English (United States)"
}
]
},
"preferred": true
}
]
}
assets/requirements.txt# Core FHIR libraries fhir.resources>=7.0.0 fhir-package-loader>=1.0.0 pydantic>=2.0.0 # Web framework fastapi>=0.100.0 uvicorn[standard]>=0.23.0 # HTTP client httpx>=0.24.0 requests>=2.31.0 # Validation and parsing jsonschema>=4.19.0 python-dateutil>=2.8.2 email-validator>=2.0.0 # Database (choose based on your needs) sqlalchemy>=2.0.0 alembic>=1.11.0 asyncpg>=0.28.0 # PostgreSQL aiomysql>=0.2.0 # MySQL aiosqlite>=0.19.0 # SQLite # Authentication/Authorization pyjwt[crypto]>=2.8.0 cryptography>=41.0.0 passlib[bcrypt]>=1.7.4 # Configuration python-dotenv>=1.0.0 pydantic-settings>=2.0.0 # Testing pytest>=7.4.0 pytest-asyncio>=0.21.0 pytest-cov>=4.1.0 httpx-test>=0.24.0 # Development tools black>=23.7.0 isort>=5.12.0 flake8>=6.0.0 mypy>=1.5.0 # Logging and monitoring structlog>=23.1.0 prometheus-client>=0.17.0 # Optional: Additional FHIR tools # fhirclient>=4.1.0 # fhirpy>=1.4.0 # fhir-parser>=0.3.0
references/fsh_ig_authoring.md# FHIR Shorthand (FSH) & IG Authoring Reference
Comprehensive reference for authoring FHIR Implementation Guides using FSH (FHIR Shorthand), SUSHI, and GoFSH.
## Toolchain Overview
| Tool | Purpose | Install |
|------|---------|---------|
| **SUSHI** | FSH compiler → FHIR JSON | `npm install -g fsh-sushi` |
| **GoFSH** | FHIR JSON → FSH converter | `npm install -g gofsh` |
| **IG Publisher** | Produces the IG website from SUSHI output | Downloaded via `_updatePublisher.sh` |
Prerequisites: Node.js LTS. Verify with `node --version` and `npm --version`.
---
## SUSHI CLI
```bash
# Initialize a new IG project scaffold
sushi init
# Build FSH → FHIR JSON (run from project root)
sushi build
# Build with options
sushi build . --snapshot # Include full StructureDefinition snapshot
sushi build . --log-level debug # Verbose output
sushi build . --out ./output # Custom output directory
sushi build . --preprocessed # Debug: dump resolved aliases/rulesets
# Other commands
sushi --version
sushi help
sushi help build
sushi update-dependencies # Bump deps to latest
```
Output lands in `fsh-generated/resources/` as `{ResourceType}-{id}.json` files. **Warning: SUSHI deletes and regenerates this folder on every run.**
---
## GoFSH CLI
Convert existing FHIR JSON conformance artifacts to FSH.
```bash
# Basic conversion (input dir contains JSON resources)
gofsh ./path/to/fhir-resources
# With dependencies beyond R4
gofsh ./definitions -d hl7.fhir.us.core@3.1.0 -d hl7.fhir.uv.ips@1.0.0
# Load existing aliases file to reuse
gofsh ./definitions --alias-file ./aliases.fsh
# Validate round-trip (GoFSH → SUSHI compare)
gofsh ./definitions --fshing-trip
# Output styles
gofsh ./definitions --style file-per-definition # default: one file per definition
gofsh ./definitions --style group-by-fsh-type # grouped by FSH entity type
gofsh ./definitions --style group-by-profile # profile + related items together
gofsh ./definitions --style single-file # everything in one file
# Indented rule output (cleaner for complex profiles)
gofsh ./definitions --indent
# Control InstanceOf derivation from meta.profile
gofsh ./definitions --meta-profile only-one # default
gofsh ./definitions --meta-profile first
gofsh ./definitions --meta-profile none
```
GoFSH outputs an `input/fsh/` directory, `index.txt`, and a starter `sushi-config.yaml`.
---
## Project Structure
### Minimal FSH-only project
```
my-project/
├── sushi-config.yaml
└── input/
└── fsh/
├── profiles.fsh
├── extensions.fsh
├── valuesets.fsh
└── instances.fsh
```
### Full IG Publisher-compatible project
```
my-project/
├── sushi-config.yaml
├── ig.ini # Required for IG Publisher
├── package-list.json # IG version history
├── _genonce.sh / _genonce.bat # Run IG Publisher
├── _updatePublisher.sh / .bat # Download latest IG Publisher
├── .gitignore
├── input/
│ ├── fsh/ # All .fsh source files
│ │ ├── profiles.fsh
│ │ ├── extensions.fsh
│ │ ├── valuesets.fsh
│ │ ├── codesystems.fsh
│ │ ├── instances.fsh
│ │ └── aliases.fsh # Centralize alias definitions
│ ├── pagecontent/ # Narrative markdown/XML pages
│ │ ├── index.md # Home page
│ │ ├── 1_background.md # Numbered = TOC order
│ │ ├── {resource-id}-intro.md # Injected before artifact def
│ │ └── {resource-id}-notes.md # Injected after artifact def
│ ├── images/ # Images, PDFs, spreadsheets
│ │ └── diagram.png
│ ├── includes/
│ │ └── menu.xml # Custom navigation menu
│ └── ignoreWarnings.txt # Suppress IG Publisher QA warnings
├── sushi-ignoreErrors.txt # Suppress SUSHI errors
├── sushi-ignoreWarnings.txt # Suppress SUSHI warnings
└── fsh-generated/ # SUSHI output (do not edit manually)
└── resources/
├── StructureDefinition-my-profile.json
├── ValueSet-my-valueset.json
└── ImplementationGuide-my.ig.json
```
---
## sushi-config.yaml Reference
```yaml
# --- Required fields ---
id: hl7.fhir.us.example # Package identifier
canonical: http://hl7.org/fhir/us/example # Base canonical URL
name: ExampleIG # No spaces
status: draft # draft | active | retired | unknown
version: 0.1.0
fhirVersion: 4.0.1 # 4.0.1 | 5.0.0 | etc.
# --- Strongly recommended ---
title: "Example Implementation Guide"
description: "An example IG for demonstration purposes."
copyrightYear: 2024+
releaseLabel: ci-build # ci-build | STU1 | Normative 1 | etc.
license: CC0-1.0 # SPDX identifier
# --- Publisher / Contact ---
publisher:
name: HL7 International
url: http://hl7.org
email: fhir@lists.hl7.org
contact:
- name: Jane Smith
telecom:
- system: email
value: jane@example.org
# --- Jurisdiction ---
jurisdiction: urn:iso:std:iso:3166#US "United States"
# --- Dependencies ---
dependencies:
hl7.fhir.us.core: 6.1.0
hl7.terminology.r4: 5.5.0
# Special versions:
# dev → local FHIR cache
# current → latest CI build
# current$branch → specific branch CI build
# Advanced with id alias for FSH use:
hl7.fhir.us.core:
id: uscore
uri: http://hl7.org/fhir/us/core/ImplementationGuide/hl7.fhir.us.core
version: 6.1.0
# --- Global profiles (applied to all resources of type) ---
global:
Patient: http://hl7.org/fhir/us/core/StructureDefinition/us-core-patient
# --- Navigation menu ---
menu:
Home: index.html
Background: background.html
Artifacts:
Profiles: artifacts.html#structures-resource-profiles
Extensions: artifacts.html#structures-extension-definitions
Value Sets: artifacts.html#terminology-value-sets
Downloads: downloads.html
# --- Pages ---
pages:
index.md:
title: Home
background.md:
title: Background
downloads.md:
title: Downloads and Links
# --- IG Publisher parameters ---
parameters:
show-inherited-invariants: false
excludettl: true
validation: [allow-any-extensions, no-broken-links]
# --- Resource overrides ---
resources:
Patient/example-patient:
name: Example Patient
description: "An example patient resource"
exampleBoolean: true # R4
# isExample: true # R5
StructureDefinition/omit-this: omit # Exclude from IG
# --- Groups (organize artifacts) ---
groups:
PatientGroup:
name: Patient Profiles
description: Profiles related to the Patient resource
resources:
- StructureDefinition/my-patient-profile
# --- Instance generation options ---
instanceOptions:
setMetaProfile: always # always | never | inline-only | standalone-only
setId: always # always | standalone-only
manualSliceOrdering: false
# --- FSH Only mode (no IG processing) ---
# FSHOnly: true
```
---
## FSH Language Reference
### Aliases
Declare at top of any `.fsh` file. By convention, prefix with `$`.
```
Alias: $LNC = http://loinc.org
Alias: $SCT = http://snomed.info/sct
Alias: $UCUM = http://unitsofmeasure.org
Alias: $V2 = http://terminology.hl7.org/CodeSystem/v2-0203
Alias: $USCoreRace = http://hl7.org/fhir/us/core/StructureDefinition/us-core-race
```
### Comments
```
// Single-line comment
/* Multi-line
block comment */
```
### Triple-Quoted Strings
For multi-line descriptions/purposes:
```
* ^purpose = """
* Intended for workflows where:
* this happens; or
* that happens
"""
```
---
## Entity Types
### Profile
```
Profile: MyPatientProfile
Parent: Patient // or URL or alias
Id: my-patient-profile
Title: "My Patient Profile"
Description: "A constrained Patient for our IG"
* identifier 1..* MS
* name 1..* MS
* birthDate MS
* gender 1..1 MS
* gender from http://hl7.org/fhir/ValueSet/administrative-gender (required)
```
### Extension
**Simple (value-bearing):**
```
Extension: PatientReligion
Id: patient-religion
Title: "Patient Religion"
Description: "The religious affiliation of the patient"
Context: Patient
* value[x] only CodeableConcept
* value[x] from ReligionValueSet (extensible)
```
**Complex (sub-extensions):**
```
Extension: USCoreEthnicityExtension
Id: us-core-ethnicity
Title: "US Core Ethnicity Extension"
Context: Patient, RelatedPerson, Practitioner, Person
* extension contains
ombCategory 0..1 MS and
detailed 0..* and
text 1..1 MS
* extension[ombCategory].value[x] only Coding
* extension[ombCategory].value[x] from OmbEthnicityCategories (required)
* extension[ombCategory] ^short = "Hispanic or Latino|Not Hispanic or Latino"
* extension[detailed].value[x] only Coding
* extension[detailed].value[x] from DetailedEthnicity (extensible)
* extension[text].value[x] only string
* extension[text] ^short = "Ethnicity Text"
```
**Context syntax:**
```
Context: Patient // simple element
Context: Patient.contact.telecom // nested element
Context: "Patient.contact where use = 'home'" // FHIRPath
Context: Patient, RelatedPerson, Practitioner // multiple
```
### Instance
```
Instance: JaneDoe
InstanceOf: MyPatientProfile
Title: "Jane Doe"
Description: "Example patient Jane Doe"
Usage: #example
* identifier[0].system = "http://hospital.example.org/mrns"
* identifier[0].value = "12345"
* name[0].family = "Doe"
* name[0].given[0] = "Jane"
* birthDate = 1970-01-01
* gender = #female
* extension[patient-religion].valueCodeableConcept = $SCT#160538000 "Christian"
```
Usage codes: `#example` (default), `#definition` (appears on own IG page), `#inline` (contained only).
### ValueSet
```
ValueSet: MyConditionStatusVS
Id: my-condition-status
Title: "My Condition Status Value Set"
Description: "Status codes for conditions in our IG"
* include codes from system $SCT where concept is-a #404684003 // Clinical finding
* include $V3#active "Active"
* include $V3#inactive "Inactive"
* exclude $SCT#74964007 "Other"
```
### CodeSystem
```
CodeSystem: MyCustomCodes
Id: my-custom-codes
Title: "My Custom Code System"
Description: "Codes defined for this IG"
* #pending "Pending" "Awaiting review"
* #approved "Approved" "Formally approved"
* #rejected "Rejected" "Not approved"
```
Hierarchical codes:
```
CodeSystem: AnatomyCS
* #body "Body" "The body"
* #head "Head" "The head"
* #eye "Eye" "The eye"
* #arm "Arm" "The arm"
```
### Invariant
```
Invariant: my-inv-1
Description: "If category is 'vital-signs', value must be present"
Severity: #error
Expression: "category.coding.code = 'vital-signs' implies value.exists()"
```
Reference in a Profile:
```
* obeys my-inv-1
```
### Mapping
```
Mapping: MyPatientToV2
Source: MyPatientProfile
Target: "http://hl7.org/v2"
Id: my-patient-v2-map
Title: "Patient to HL7 v2 Mapping"
* -> "PID"
* name -> "PID-5 (family), PID-9 (given)"
* birthDate -> "PID-7"
* gender -> "PID-8"
```
### Logical Model
```
Logical: ClinicalAssessment
Id: clinical-assessment
Title: "Clinical Assessment"
Description: "Logical model for a clinical assessment document"
* assessmentDate 1..1 dateTime "Date of assessment" "When performed"
* assessor 1..* Reference(Practitioner) "Assessor" "Who performed"
* findings 0..* BackboneElement "Findings" "Clinical findings"
* code 1..1 CodeableConcept "Finding code" "What was found"
* value[x] 0..1 string or Quantity "Finding value" "Result"
```
### RuleSet (Reusable Rules)
**Simple RuleSet:**
```
RuleSet: PublicationMetadata
* ^status = #active
* ^experimental = false
* ^publisher = "My Organization"
* ^date = "2024-01-01"
// Usage:
Profile: MyProfile
Parent: Observation
* insert PublicationMetadata
```
**Parameterized RuleSet:**
```
RuleSet: SetContext(contextPath)
* ^context[+].type = #element
* ^context[=].expression = "{contextPath}"
// Usage:
Extension: MyExtension
* insert SetContext(Patient)
* insert SetContext(Practitioner)
```
---
## Rules Reference
### Cardinality
```
* element 0..1 // Optional, at most one
* element 1..* // At least one, unbounded
* element 1..1 // Exactly one
* element 0..* // Optional, unbounded (default)
```
### Flags
```
* element MS // Must Support
* element SU // Summary (include in _summary)
* element ?! // Is Modifier
* element N // Normative
* element TU // Trial Use
* element D // Draft
// Combine:
* element 1..* MS SU
```
### Type Constraint
```
* value[x] only Quantity
* value[x] only Quantity or CodeableConcept
* performer only Reference(Practitioner or Organization)
```
### Binding
```
* code from http://hl7.org/fhir/ValueSet/observation-codes (required)
* status from ObservationStatusVS (extensible)
* category from $ObsCat (preferred)
* method from MethodVS (example)
```
### Assignment (Fixed / Default Values)
```
// Code
* status = #final
// Coding with system
* code = $LNC#29463-7 "Body weight"
// Coding with alias
* code = $SCT#27113001 "Body weight"
// Quantity
* valueQuantity = 70 'kg' "kg"
// Boolean, String, Integer, Decimal
* active = true
* name = "Default Name"
* multipleBirthInteger = 2
* score = 9.5
// Reference to profile type
* subject only Reference(MyPatientProfile)
// Canonical
* instantiatesCanonical = Canonical(MyPlanDefinition)
```
### Slicing (Contains)
```
// Slice a repeating element
* category contains
VSCat 1..1 MS
// Set discriminator via caret
* category ^slicing.discriminator[0].type = #pattern
* category ^slicing.discriminator[0].path = "$this"
* category ^slicing.rules = #open
// Constrain the slice
* category[VSCat].coding 1..* MS
* category[VSCat].coding.system 1..1 MS
* category[VSCat].coding.system = "http://terminology.hl7.org/CodeSystem/observation-category"
* category[VSCat].coding.code 1..1 MS
* category[VSCat].coding.code = #vital-signs
```
### Extension Slicing (Profile adding extensions)
```
// Add a standalone extension to a profile
* extension contains
$USCoreRace named race 0..1 MS and
$USCoreEthnicity named ethnicity 0..1 MS
* extension[race] ^short = "US Core Race Extension"
```
### Caret Rules (Metadata / ElementDefinition properties)
```
// StructureDefinition metadata
* ^url = "http://example.org/StructureDefinition/my-profile"
* ^version = "1.0.0"
* ^experimental = false
* ^purpose = "Defines constraints for our use case"
// ElementDefinition properties
* code ^binding.description = "Codes for conditions"
* name ^example[0].label = "Example name"
* name ^example[0].valueHumanName.family = "Smith"
// Self element (the root)
* . ^short = "Profile root short description"
* . ^comment = "Extended comment on the profile root"
```
### ValueSet Include/Exclude Rules
```
// Include entire code system
* include codes from system http://loinc.org
// Include specific concepts
* $SCT#73211009 "Diabetes mellitus"
* $SCT#44054006 "Diabetes mellitus type 2"
// Include by filter
* include codes from system $SCT where concept is-a #73211009
// Include another ValueSet
* include codes from valueset http://hl7.org/fhir/ValueSet/condition-code
// Exclude
* exclude codes from system http://loinc.org where STATUS = DEPRECATED
```
---
## Path Grammar
| Pattern | Example | Meaning |
|---------|---------|---------|
| Simple element | `status` | Direct child element |
| Nested | `name.family` | Nested element |
| Choice type | `valueQuantity` | `value[x]` as Quantity |
| Array index | `name[0]` | First item (zero-based) |
| Soft index (increment) | `name[+]` | Next slot |
| Soft index (same) | `name[=]` | Same slot as previous |
| Slice | `component[respirationScore]` | Named slice |
| Slice + index | `component[respirationScore][1]` | Second item in slice |
| Extension | `extension[race]` | Extension slice by name |
| Extension by URL | `extension[http://hl7.org/fhir/us/core/StructureDefinition/us-core-race]` | By full URL |
| Reference target | `performer[Practitioner]` | Specific Reference type |
| Caret (metadata) | `^experimental` | StructureDefinition property |
| Element metadata | `status ^short` | ElementDefinition property |
---
## Code / Coding / Quantity Syntax
```
// Simple code (no system)
#active
#"code with spaces"
// Coding: system#code "display"
http://loinc.org#29463-7 "Body Weight"
$LNC#29463-7 "Body Weight"
// Coding with version
http://loinc.org|2.73#29463-7
// Quantity (UCUM)
70.5 'kg' "kg"
98.6 '[degF]' "F"
// Quantity (non-UCUM system)
155 http://unitsofmeasure.org#C48531 "Pound"
```
---
## Complete Worked Example
```
// aliases.fsh
Alias: $LNC = http://loinc.org
Alias: $SCT = http://snomed.info/sct
Alias: $UCUM = http://unitsofmeasure.org
// profiles.fsh
Profile: VitalSignsObservation
Parent: http://hl7.org/fhir/StructureDefinition/vitalsigns
Id: vital-signs-observation
Title: "Vital Signs Observation"
Description: "Constrained vital signs profile for our IG"
* insert PublicationMetadata
* status MS
* category MS
* code MS
* subject 1..1 MS
* subject only Reference(MyPatientProfile)
* effective[x] 1..1 MS
* effective[x] only dateTime or Period
* value[x] MS
* value[x] only Quantity
* valueQuantity.value 1..1 MS
* valueQuantity.unit 1..1 MS
* valueQuantity.system 1..1 MS
* valueQuantity.system = $UCUM
* valueQuantity.code 1..1 MS
// rulesets.fsh
RuleSet: PublicationMetadata
* ^status = #active
* ^experimental = false
* ^publisher = "My Organization"
// invariants.fsh
Invariant: vs-quantity-present
Description: "Value quantity must have value and unit when present"
Severity: #error
Expression: "valueQuantity.exists() implies (valueQuantity.value.exists() and valueQuantity.unit.exists())"
// instances.fsh
Instance: BloodPressureExample
InstanceOf: VitalSignsObservation
Title: "Blood Pressure Example"
Usage: #example
* status = #final
* category[VSCat].coding.system = "http://terminology.hl7.org/CodeSystem/observation-category"
* category[VSCat].coding.code = #vital-signs
* code = $LNC#85354-9 "Blood pressure systolic and diastolic"
* subject = Reference(JaneDoe)
* effectiveDateTime = "2024-01-15T10:30:00Z"
* component[+].code = $LNC#8480-6 "Systolic blood pressure"
* component[=].valueQuantity = 120 'mm[Hg]' "mmHg"
* component[+].code = $LNC#8462-4 "Diastolic blood pressure"
* component[=].valueQuantity = 80 'mm[Hg]' "mmHg"
```
---
## Common IG Authoring Workflows
### Start a new IG from scratch
```bash
mkdir my-ig && cd my-ig
sushi init # Interactive scaffold generator
# Edit sushi-config.yaml
# Write .fsh files in input/fsh/
sushi build # Compile FSH → JSON
./_genonce.sh # Run IG Publisher (first run downloads it)
```
### Migrate existing FHIR JSON to FSH
```bash
# Point GoFSH at existing JSON conformance artifacts
gofsh ./existing-ig/resources \
-d hl7.fhir.us.core@6.1.0 \
--style group-by-profile \
--indent \
--fshing-trip # Validate round-trip accuracy
# Review output in ./fsh-output/input/fsh/
# Then move to your project's input/fsh/
```
### Debug a failing build
```bash
sushi build --log-level debug
sushi build --preprocessed # See resolved aliases/rulesets in _preprocessed/
```
### Update IG Publisher
```bash
./_updatePublisher.sh # Downloads latest jar
```
---
## Best Practices
### File Organization
- `aliases.fsh` — All `Alias:` declarations (one place to update URLs)
- `profiles.fsh` — Profile definitions
- `extensions.fsh` — Extension definitions
- `valuesets.fsh` — ValueSet and CodeSystem definitions
- `instances.fsh` — Example instances (Usage: #example)
- `rulesets.fsh` — Shared RuleSets
- `invariants.fsh` — Invariant definitions
### Naming Conventions
- **Item names** (Profile, Extension, etc.): `PascalCase` (e.g., `MyPatientProfile`)
- **Item IDs**: `kebab-case`, max 64 chars (e.g., `my-patient-profile`)
- **Slice names**: `lowerCamelCase` (e.g., `respirationScore`)
- **Aliases**: Prefix with `$` (e.g., `$LNC`, `$SCT`)
- **RuleSets**: `PascalCase`, descriptive (e.g., `PublicationMetadata`)
### Slicing Order
Always define slices (Contains rule) **before** constraining individual slices:
```
// 1. First: declare the slices
* component contains
systolic 1..1 MS and
diastolic 1..1 MS
// 2. Then: constrain each slice
* component[systolic].code = $LNC#8480-6
* component[diastolic].code = $LNC#8462-4
```
### Extension Order
Always declare extension slices before constraining sub-elements:
```
// 1. Declare
* extension contains
ombCategory 0..1 MS and
text 1..1 MS
// 2. Constrain
* extension[ombCategory].value[x] only Coding
* extension[text].value[x] only string
```
### Must Support Strategy
Flag elements as MS based on your IG's definition. Be explicit:
```
* identifier 1..* MS
* identifier.system 1..1 MS
* identifier.value 1..1 MS
```
### Canonical URLs
Keep consistent with your `canonical` in sushi-config.yaml. SUSHI automatically generates canonical URLs using `{canonical}/StructureDefinition/{id}`.
### Dependencies
Pin exact versions for reproducible builds. Use `current` only for development against unreleased IGs.
---
## Key FSH Spec Links
- Full Spec: https://build.fhir.org/ig/HL7/fhir-shorthand/
- Language Reference: https://build.fhir.org/ig/HL7/fhir-shorthand/reference.html
- SUSHI Docs: https://fshschool.org/docs/sushi/
- GoFSH Docs: https://fshschool.org/docs/gofsh/
- FSH School (tutorials): https://fshschool.org/
references/search_parameters.md# FHIR Search Parameters and API Patterns Reference
This document provides comprehensive guidance on implementing FHIR search functionality and API patterns.
## FHIR Search Parameter Types
### 1. String Parameters
Search on string fields with prefix matching and case-insensitive behavior.
**Examples:**
- `GET /Patient?name=john` - Find patients with "john" in any name component
- `GET /Patient?family=doe` - Search by family name
- `GET /Organization?name:exact=ACME Corp` - Exact string match
**Modifiers:**
- `:exact` - Exact case-sensitive match
- `:contains` - Substring search
- `:missing` - Search for missing/present values
### 2. Token Parameters
Search on coded values, identifiers, and boolean fields.
**Format:** `[system]|[code]`
**Examples:**
- `GET /Patient?identifier=123456` - Patient with identifier value
- `GET /Patient?identifier=http://hospital.org|123456` - With specific system
- `GET /Observation?code=http://loinc.org|55284-4` - Specific LOINC code
- `GET /Patient?active=true` - Boolean search
**Modifiers:**
- `:text` - Search in display text
- `:not` - Negation
- `:above` / `:below` - Code hierarchy navigation
### 3. Date/DateTime Parameters
Search on date and dateTime fields with precision handling.
**Examples:**
- `GET /Patient?birthdate=1990` - Born in 1990
- `GET /Patient?birthdate=ge2020-01-01` - Born on or after Jan 1, 2020
- `GET /Observation?date=2023-03-15` - Observations on specific date
**Prefixes:**
- `eq` - Equal (default)
- `ne` - Not equal
- `gt` - Greater than
- `lt` - Less than
- `ge` - Greater or equal
- `le` - Less or equal
- `sa` - Starts after
- `eb` - Ends before
- `ap` - Approximately
### 4. Number Parameters
Search on numeric values with precision and range support.
**Examples:**
- `GET /Observation?value-quantity=5.4` - Value equals 5.4
- `GET /Observation?value-quantity=gt100` - Value greater than 100
- `GET /RiskAssessment?probability=lt0.8` - Probability less than 0.8
### 5. Reference Parameters
Search by references to other resources.
**Examples:**
- `GET /Observation?subject=Patient/123` - References specific patient
- `GET /Observation?subject:Patient=123` - Type-specific reference
- `GET /DiagnosticReport?result=Observation/456` - References observation
**Chaining:**
- `GET /Observation?subject.name=smith` - Chain to referenced resource
- `GET /DiagnosticReport?subject:Patient.birthdate=1990` - Chain with type
### 6. Quantity Parameters
Search on Quantity values with units and comparisons.
**Format:** `[prefix][number]|[system]|[code]`
**Examples:**
- `GET /Observation?value-quantity=5.4|http://unitsofmeasure.org|mg`
- `GET /Observation?value-quantity=gt100||mg` - Greater than 100mg
- `GET /Observation?value-quantity=ap5.4` - Approximately 5.4
### 7. Composite Parameters
Search on multiple related fields simultaneously.
**Examples:**
- `GET /Observation?code-value-quantity=http://loinc.org|33747-0$gt5.4|http://unitsofmeasure.org|mmol/L`
- `GET /Observation?component-code-value-quantity=...` - Component values
## Search Modifiers
### Universal Modifiers
- `:missing=true|false` - Search for missing/present values
- `:exact` - Exact string matching
- `:contains` - Substring search
- `:text` - Search in human-readable text
### Reference Modifiers
- `:identifier` - Search by identifier instead of ID
- `:[type]` - Type-specific reference search
- `:iterate` - Follow reference chains
### String Modifiers
- `:exact` - Case-sensitive exact match
- `:contains` - Case-insensitive substring
## Search Result Parameters
### Pagination
```
GET /Patient?_count=20&_offset=40
GET /Patient?_page=3&_count=20
```
### Include/Reverse Include
```
GET /DiagnosticReport?_include=DiagnosticReport:subject
GET /Patient?_revinclude=Observation:subject
GET /Patient?_include=Patient:organization&_include:iterate=Organization:partof
```
### Sorting
```
GET /Patient?_sort=family
GET /Patient?_sort=birthdate,-family # Multiple sort, desc family
GET /Observation?_sort=date:desc
```
### Element Selection
```
GET /Patient?_elements=id,name,birthDate
GET /Patient?_summary=true
GET /Patient?_summary=text
GET /Patient?_summary=data
```
### Format Control
```
GET /Patient?_format=json
GET /Patient?_format=xml
GET /Patient?_format=application/fhir+json
```
## Advanced Search Patterns
### Composite Searches
Combine multiple parameters with AND logic:
```
GET /Patient?name=smith&birthdate=1990&active=true
```
### OR Logic with Multiple Values
```
GET /Patient?name=smith,jones,wilson
GET /Observation?code=http://loinc.org|12345,http://loinc.org|67890
```
### Complex Date Ranges
```
GET /Observation?date=ge2023-01-01&date=lt2023-02-01
GET /Patient?birthdate=ge1990&birthdate=le1995
```
### Chained Searches
```
GET /DiagnosticReport?subject.gender=female
GET /Observation?subject:Patient.name=smith
GET /MedicationRequest?patient.identifier=123456
```
### Reverse Chaining
```
GET /Patient?_has:Observation:subject:code=http://loinc.org|55284-4
GET /Patient?_has:Condition:subject:category=problem-list-item
```
## Search Implementation Patterns
### Search Parameter Processing
```python
def process_search_params(resource_type: str, params: dict) -> dict:
"""Convert FHIR search parameters to database query"""
query = {}
for param_name, values in params.items():
# Handle modifiers
base_param, modifier = parse_modifier(param_name)
# Get parameter definition
search_param = get_search_parameter(resource_type, base_param)
if not search_param:
continue
# Process based on parameter type
if search_param['type'] == 'string':
query[base_param] = process_string_search(values, modifier)
elif search_param['type'] == 'token':
query[base_param] = process_token_search(values, modifier)
elif search_param['type'] == 'date':
query[base_param] = process_date_search(values, modifier)
# ... other types
return query
```
### Bundle Search Results
```python
def create_search_bundle(resources: list, total: int, params: dict) -> dict:
"""Create FHIR Bundle for search results"""
return {
'resourceType': 'Bundle',
'type': 'searchset',
'total': total,
'link': [
{
'relation': 'self',
'url': build_search_url(params)
},
# Add next/prev links for pagination
],
'entry': [
{
'resource': resource,
'search': {'mode': 'match'}
} for resource in resources
]
}
```
### Compartment-based Search
```python
# Patient compartment search
GET /Patient/123/Observation # All observations for patient 123
GET /Patient/123/DiagnosticReport
GET /Patient/123/* # All resources in patient compartment
def get_compartment_resources(compartment_type: str, id: str, resource_type: str = None):
"""Get resources within a compartment"""
if compartment_type == 'Patient':
return get_patient_compartment_resources(id, resource_type)
elif compartment_type == 'Encounter':
return get_encounter_compartment_resources(id, resource_type)
# ... other compartments
```
## Error Handling
### Invalid Search Parameters
```json
{
"resourceType": "OperationOutcome",
"issue": [{
"severity": "error",
"code": "invalid",
"details": {
"text": "Unknown search parameter: invalidParam"
},
"location": ["invalidParam"]
}]
}
```
### Search Parameter Validation
```python
def validate_search_parameters(resource_type: str, params: dict) -> list:
"""Validate search parameters and return issues"""
issues = []
for param_name in params:
base_param, modifier = parse_modifier(param_name)
# Check if parameter exists
if not search_parameter_exists(resource_type, base_param):
issues.append({
'severity': 'error',
'code': 'not-supported',
'details': {'text': f'Unknown search parameter: {base_param}'},
'location': [param_name]
})
# Validate modifier
if modifier and not modifier_supported(base_param, modifier):
issues.append({
'severity': 'error',
'code': 'not-supported',
'details': {'text': f'Unsupported modifier: {modifier}'},
'location': [param_name]
})
return issues
```
## Performance Optimization
### Indexed Search Parameters
```sql
-- Database indexes for common FHIR searches
CREATE INDEX idx_patient_family ON patient_names(family);
CREATE INDEX idx_patient_given ON patient_names(given);
CREATE INDEX idx_patient_birthdate ON patients(birth_date);
CREATE INDEX idx_observation_code ON observations(code_system, code_value);
CREATE INDEX idx_observation_subject ON observations(subject_reference);
CREATE INDEX idx_observation_date ON observations(effective_datetime);
```
### Search Parameter Caching
```python
@lru_cache(maxsize=1000)
def get_search_parameter(resource_type: str, param_name: str) -> dict:
"""Cache search parameter definitions"""
return load_search_parameter_from_spec(resource_type, param_name)
```
### Batch Search Processing
```python
async def batch_search(searches: list) -> dict:
"""Process multiple searches in a batch"""
results = {}
# Group searches by resource type for optimization
grouped = group_searches_by_type(searches)
for resource_type, type_searches in grouped.items():
type_results = await process_resource_searches(resource_type, type_searches)
results.update(type_results)
return results
```
## Testing Search Implementation
### Search Parameter Tests
```python
def test_string_search():
# Test basic string search
result = search_patients({'name': 'john'})
assert any('john' in name.lower() for patient in result for name in patient.get_names())
# Test exact modifier
result = search_patients({'name:exact': 'John'})
assert all('John' in patient.get_display_names() for patient in result)
def test_token_search():
# Test system|code format
result = search_observations({'code': 'http://loinc.org|55284-4'})
assert all(obs.code.system == 'http://loinc.org' and obs.code.code == '55284-4'
for obs in result)
def test_date_search():
# Test date range
result = search_patients({'birthdate': 'ge1990-01-01&birthdate=lt2000-01-01'})
assert all(1990 <= get_birth_year(patient) < 2000 for patient in result)
```
## Integration with FHIR Servers
### Popular FHIR Server Search APIs
**HAPI FHIR:**
- Full SearchParameter support
- Custom search parameter definitions
- Subscription support
**Microsoft FHIR Server:**
- Core search parameters
- Custom search parameters via SearchParameter resources
- Azure-optimized queries
**Google FHIR Store:**
- Standard FHIR search
- BigQuery integration for analytics
- Custom search with views
**AWS HealthLake:**
- FHIR R4 search support
- Natural language query (NLQ)
- Integrated analytics
references/smart_on_fhir.md# SMART on FHIR Implementation Reference
This document provides comprehensive guidance for implementing SMART on FHIR applications and authorization flows.
## SMART App Launch Framework
### Authorization Code Flow
#### 1. Discovery
The SMART app discovers the authorization and token endpoints from the FHIR server's well-known configuration.
```javascript
// Discover SMART configuration
async function discoverSmartConfig(fhirBaseUrl) {
const configUrl = `${fhirBaseUrl}/.well-known/smart_configuration`;
try {
const response = await fetch(configUrl);
const config = await response.json();
return {
authorizationEndpoint: config.authorization_endpoint,
tokenEndpoint: config.token_endpoint,
capabilities: config.capabilities || [],
scopesSupported: config.scopes_supported || [],
responseTypesSupported: config.response_types_supported || []
};
} catch (error) {
// Fallback to CapabilityStatement discovery
return await discoverFromCapabilityStatement(fhirBaseUrl);
}
}
async function discoverFromCapabilityStatement(fhirBaseUrl) {
const capabilityUrl = `${fhirBaseUrl}/metadata`;
const response = await fetch(capabilityUrl);
const capability = await response.json();
const smartExtension = capability.rest?.[0]?.security?.extension?.find(
ext => ext.url === 'http://fhir-registry.smarthealthit.org/StructureDefinition/oauth-uris'
);
if (smartExtension) {
const authExt = smartExtension.extension.find(e => e.url === 'authorize');
const tokenExt = smartExtension.extension.find(e => e.url === 'token');
return {
authorizationEndpoint: authExt?.valueUri,
tokenEndpoint: tokenExt?.valueUri,
capabilities: [],
scopesSupported: [],
responseTypesSupported: ['code']
};
}
throw new Error('SMART configuration not found');
}
```
#### 2. Authorization Request
```javascript
function buildAuthorizationUrl(smartConfig, appConfig) {
const authUrl = new URL(smartConfig.authorizationEndpoint);
const state = generateSecureRandom(32);
const codeVerifier = generateCodeVerifier();
const codeChallenge = generateCodeChallenge(codeVerifier);
// Store for later use
sessionStorage.setItem('smart_state', state);
sessionStorage.setItem('code_verifier', codeVerifier);
authUrl.searchParams.set('response_type', 'code');
authUrl.searchParams.set('client_id', appConfig.clientId);
authUrl.searchParams.set('redirect_uri', appConfig.redirectUri);
authUrl.searchParams.set('launch', appConfig.launchToken); // For EHR launch
authUrl.searchParams.set('scope', appConfig.scopes.join(' '));
authUrl.searchParams.set('state', state);
authUrl.searchParams.set('aud', appConfig.fhirBaseUrl);
// PKCE parameters
authUrl.searchParams.set('code_challenge', codeChallenge);
authUrl.searchParams.set('code_challenge_method', 'S256');
return authUrl.toString();
}
function generateCodeVerifier() {
const array = new Uint8Array(32);
crypto.getRandomValues(array);
return base64urlEncode(array);
}
function generateCodeChallenge(verifier) {
const encoder = new TextEncoder();
const data = encoder.encode(verifier);
return crypto.subtle.digest('SHA-256', data).then(hash => base64urlEncode(new Uint8Array(hash)));
}
```
#### 3. Token Exchange
```javascript
async function exchangeCodeForToken(authCode, smartConfig, appConfig) {
const codeVerifier = sessionStorage.getItem('code_verifier');
const tokenRequest = {
grant_type: 'authorization_code',
client_id: appConfig.clientId,
code: authCode,
redirect_uri: appConfig.redirectUri,
code_verifier: codeVerifier
};
const response = await fetch(smartConfig.tokenEndpoint, {
method: 'POST',
headers: {
'Content-Type': 'application/x-www-form-urlencoded',
'Accept': 'application/json'
},
body: new URLSearchParams(tokenRequest)
});
if (!response.ok) {
throw new Error(`Token exchange failed: ${response.statusText}`);
}
const tokenResponse = await response.json();
return {
accessToken: tokenResponse.access_token,
tokenType: tokenResponse.token_type || 'Bearer',
expiresIn: tokenResponse.expires_in,
refreshToken: tokenResponse.refresh_token,
scope: tokenResponse.scope,
patient: tokenResponse.patient, // Patient context
encounter: tokenResponse.encounter, // Encounter context
fhirUser: tokenResponse.fhirUser // Practitioner context
};
}
```
## SMART Scopes
### Clinical Scopes
```javascript
const SMART_SCOPES = {
// Patient scopes
PATIENT_READ: 'patient/*.read',
PATIENT_WRITE: 'patient/*.write',
PATIENT_ALL: 'patient/*.*',
// Specific resource scopes
PATIENT_PATIENT_READ: 'patient/Patient.read',
PATIENT_OBSERVATION_READ: 'patient/Observation.read',
PATIENT_CONDITION_READ: 'patient/Condition.read',
PATIENT_MEDICATION_READ: 'patient/MedicationRequest.read',
// User scopes (practitioner context)
USER_READ: 'user/*.read',
USER_WRITE: 'user/*.write',
USER_ALL: 'user/*.*',
// System scopes (backend services)
SYSTEM_READ: 'system/*.read',
SYSTEM_WRITE: 'system/*.write',
SYSTEM_ALL: 'system/*.*',
// Profile scopes
OPENID: 'openid',
FHIR_USER: 'fhirUser',
PROFILE: 'profile',
// Launch context
LAUNCH: 'launch',
LAUNCH_PATIENT: 'launch/patient',
LAUNCH_ENCOUNTER: 'launch/encounter',
// Offline access
OFFLINE_ACCESS: 'offline_access'
};
function buildScopeString(resourceTypes, permissions, context = 'patient') {
return resourceTypes.map(resource =>
permissions.map(permission => `${context}/${resource}.${permission}`)
).flat().join(' ');
}
// Example usage
const scopes = [
SMART_SCOPES.LAUNCH,
SMART_SCOPES.OPENID,
SMART_SCOPES.FHIR_USER,
buildScopeString(['Patient', 'Observation', 'Condition'], ['read'], 'patient'),
SMART_SCOPES.OFFLINE_ACCESS
].join(' ');
```
## FHIR Client with SMART Authentication
```javascript
class SMARTFHIRClient {
constructor(baseUrl, tokenInfo) {
this.baseUrl = baseUrl;
this.tokenInfo = tokenInfo;
}
async request(method, path, data = null) {
const url = `${this.baseUrl}/${path}`;
const headers = {
'Authorization': `${this.tokenInfo.tokenType} ${this.tokenInfo.accessToken}`,
'Accept': 'application/fhir+json',
'Content-Type': 'application/fhir+json'
};
const options = {
method,
headers
};
if (data) {
options.body = JSON.stringify(data);
}
const response = await fetch(url, options);
if (response.status === 401) {
// Token expired, attempt refresh
if (this.tokenInfo.refreshToken) {
await this.refreshAccessToken();
// Retry request with new token
headers.Authorization = `${this.tokenInfo.tokenType} ${this.tokenInfo.accessToken}`;
return fetch(url, { ...options, headers });
} else {
throw new Error('Authentication required');
}
}
if (!response.ok) {
throw new Error(`FHIR request failed: ${response.status} ${response.statusText}`);
}
return response.json();
}
async refreshAccessToken() {
const refreshRequest = {
grant_type: 'refresh_token',
refresh_token: this.tokenInfo.refreshToken,
client_id: this.clientId
};
const response = await fetch(this.tokenEndpoint, {
method: 'POST',
headers: {
'Content-Type': 'application/x-www-form-urlencoded'
},
body: new URLSearchParams(refreshRequest)
});
const newTokenInfo = await response.json();
this.tokenInfo.accessToken = newTokenInfo.access_token;
this.tokenInfo.expiresIn = newTokenInfo.expires_in;
if (newTokenInfo.refresh_token) {
this.tokenInfo.refreshToken = newTokenInfo.refresh_token;
}
}
// FHIR operations
async read(resourceType, id) {
return this.request('GET', `${resourceType}/${id}`);
}
async search(resourceType, params = {}) {
const queryString = new URLSearchParams(params).toString();
const path = queryString ? `${resourceType}?${queryString}` : resourceType;
return this.request('GET', path);
}
async create(resource) {
return this.request('POST', resource.resourceType, resource);
}
async update(resource) {
return this.request('PUT', `${resource.resourceType}/${resource.id}`, resource);
}
async delete(resourceType, id) {
return this.request('DELETE', `${resourceType}/${id}`);
}
// Patient context operations
async getPatientContext() {
if (this.tokenInfo.patient) {
return this.read('Patient', this.tokenInfo.patient);
}
throw new Error('No patient context available');
}
async getPatientObservations(params = {}) {
if (this.tokenInfo.patient) {
return this.search('Observation', {
...params,
subject: `Patient/${this.tokenInfo.patient}`
});
}
throw new Error('No patient context available');
}
}
```
## Backend Services (System Scopes)
### Client Credentials Flow
```javascript
class SMARTBackendClient {
constructor(config) {
this.config = config;
this.accessToken = null;
this.tokenExpiry = null;
}
async authenticate() {
const now = Math.floor(Date.now() / 1000);
const exp = now + 300; // 5 minutes
// Create JWT assertion
const header = {
alg: 'RS384',
typ: 'JWT',
kid: this.config.keyId
};
const payload = {
iss: this.config.clientId,
sub: this.config.clientId,
aud: this.config.tokenEndpoint,
exp: exp,
iat: now,
jti: generateUUID()
};
const assertion = await this.signJWT(header, payload);
const tokenRequest = {
grant_type: 'client_credentials',
scope: this.config.scopes.join(' '),
client_assertion_type: 'urn:ietf:params:oauth:client-assertion-type:jwt-bearer',
client_assertion: assertion
};
const response = await fetch(this.config.tokenEndpoint, {
method: 'POST',
headers: {
'Content-Type': 'application/x-www-form-urlencoded'
},
body: new URLSearchParams(tokenRequest)
});
const tokenResponse = await response.json();
this.accessToken = tokenResponse.access_token;
this.tokenExpiry = now + tokenResponse.expires_in;
return tokenResponse;
}
async signJWT(header, payload) {
// Implementation depends on your JWT library
// Example with node-jsonwebtoken:
const jwt = require('jsonwebtoken');
return jwt.sign(payload, this.config.privateKey, {
algorithm: 'RS384',
keyid: this.config.keyId,
header: header
});
}
async ensureValidToken() {
const now = Math.floor(Date.now() / 1000);
if (!this.accessToken || now >= this.tokenExpiry - 60) {
await this.authenticate();
}
}
async request(method, path, data = null) {
await this.ensureValidToken();
const url = `${this.config.fhirBaseUrl}/${path}`;
const headers = {
'Authorization': `Bearer ${this.accessToken}`,
'Accept': 'application/fhir+json'
};
if (data) {
headers['Content-Type'] = 'application/fhir+json';
}
const response = await fetch(url, {
method,
headers,
body: data ? JSON.stringify(data) : null
});
if (!response.ok) {
throw new Error(`Request failed: ${response.status} ${response.statusText}`);
}
return response.json();
}
}
```
## Security Considerations
### Token Storage
```javascript
class SecureTokenStorage {
static store(tokenInfo, persistent = false) {
const storage = persistent ? localStorage : sessionStorage;
// Encrypt sensitive data before storage
const encrypted = this.encrypt(JSON.stringify(tokenInfo));
storage.setItem('smart_tokens', encrypted);
}
static retrieve() {
const encrypted = sessionStorage.getItem('smart_tokens') ||
localStorage.getItem('smart_tokens');
if (encrypted) {
const decrypted = this.decrypt(encrypted);
return JSON.parse(decrypted);
}
return null;
}
static clear() {
sessionStorage.removeItem('smart_tokens');
localStorage.removeItem('smart_tokens');
}
static encrypt(data) {
// Implementation depends on your crypto library
// Use Web Crypto API for browser environments
return btoa(data); // Simplified - use proper encryption
}
static decrypt(encryptedData) {
return atob(encryptedData); // Simplified - use proper decryption
}
}
```
### CSRF Protection
```javascript
function validateState(receivedState) {
const storedState = sessionStorage.getItem('smart_state');
sessionStorage.removeItem('smart_state');
if (!storedState || receivedState !== storedState) {
throw new Error('Invalid state parameter - possible CSRF attack');
}
}
```
## Testing SMART Applications
### Mock SMART Server
```javascript
class MockSMARTServer {
constructor() {
this.clients = new Map();
this.tokens = new Map();
this.patients = new Map();
}
registerClient(clientId, config) {
this.clients.set(clientId, config);
}
handleAuthorizationRequest(params) {
const clientId = params.get('client_id');
const redirectUri = params.get('redirect_uri');
const scope = params.get('scope');
const state = params.get('state');
// Validate client
const client = this.clients.get(clientId);
if (!client || !client.redirectUris.includes(redirectUri)) {
throw new Error('Invalid client or redirect URI');
}
// Generate authorization code
const code = generateSecureRandom(32);
this.tokens.set(code, {
clientId,
redirectUri,
scope,
expiresAt: Date.now() + 600000, // 10 minutes
codeChallenge: params.get('code_challenge')
});
// Redirect with code
const redirectUrl = new URL(redirectUri);
redirectUrl.searchParams.set('code', code);
redirectUrl.searchParams.set('state', state);
return redirectUrl.toString();
}
handleTokenRequest(body) {
const code = body.get('code');
const clientId = body.get('client_id');
const codeVerifier = body.get('code_verifier');
const authData = this.tokens.get(code);
if (!authData || authData.expiresAt < Date.now()) {
throw new Error('Invalid or expired authorization code');
}
// Validate PKCE
if (authData.codeChallenge) {
const challenge = generateCodeChallenge(codeVerifier);
if (challenge !== authData.codeChallenge) {
throw new Error('Invalid code verifier');
}
}
// Generate access token
const accessToken = generateSecureRandom(32);
return {
access_token: accessToken,
token_type: 'Bearer',
expires_in: 3600,
scope: authData.scope,
patient: 'example-patient-id'
};
}
}
```
## Integration Examples
### React SMART App
```jsx
import React, { useState, useEffect } from 'react';
function SMARTApp() {
const [client, setClient] = useState(null);
const [patient, setPatient] = useState(null);
const [observations, setObservations] = useState([]);
const [loading, setLoading] = useState(true);
useEffect(() => {
initializeSMART();
}, []);
async function initializeSMART() {
try {
// Check for authorization code in URL
const urlParams = new URLSearchParams(window.location.search);
const code = urlParams.get('code');
const state = urlParams.get('state');
if (code) {
// Exchange code for token
validateState(state);
const smartConfig = await discoverSmartConfig(FHIR_BASE_URL);
const tokenInfo = await exchangeCodeForToken(code, smartConfig, APP_CONFIG);
// Initialize FHIR client
const fhirClient = new SMARTFHIRClient(FHIR_BASE_URL, tokenInfo);
setClient(fhirClient);
// Load patient context
const patientData = await fhirClient.getPatientContext();
setPatient(patientData);
// Load observations
const obsData = await fhirClient.getPatientObservations({
category: 'vital-signs',
_sort: '-date'
});
setObservations(obsData.entry || []);
} else {
// Redirect to authorization
const smartConfig = await discoverSmartConfig(FHIR_BASE_URL);
const authUrl = buildAuthorizationUrl(smartConfig, APP_CONFIG);
window.location.href = authUrl;
}
} catch (error) {
console.error('SMART initialization failed:', error);
} finally {
setLoading(false);
}
}
if (loading) {
return <div>Loading...</div>;
}
return (
<div>
<h1>SMART on FHIR App</h1>
{patient && (
<div>
<h2>Patient: {patient.name?.[0]?.given?.join(' ')} {patient.name?.[0]?.family}</h2>
<p>DOB: {patient.birthDate}</p>
<p>Gender: {patient.gender}</p>
</div>
)}
<h3>Recent Vital Signs</h3>
<ul>
{observations.map((entry, index) => (
<li key={index}>
{entry.resource.code.coding[0].display}: {entry.resource.valueQuantity?.value} {entry.resource.valueQuantity?.unit}
</li>
))}
</ul>
</div>
);
}
export default SMARTApp;
```
This reference provides comprehensive patterns for implementing SMART on FHIR applications across different scenarios and environments.
references/testing_patterns.md# FHIR Testing and Quality Assurance Reference
This document provides comprehensive guidance for testing FHIR implementations, including unit tests, integration tests, validation tests, and performance testing.
## Unit Testing FHIR Resources
### Resource Validation Tests
```python
import pytest
from fhir.resources.patient import Patient
from fhir.resources.observation import Observation
from pydantic import ValidationError
import json
class TestFHIRResources:
def test_valid_patient_creation(self):
"""Test creating a valid patient resource"""
patient_data = {
"resourceType": "Patient",
"id": "test-patient",
"active": True,
"name": [
{
"family": "Doe",
"given": ["John", "Michael"]
}
],
"gender": "male",
"birthDate": "1990-01-01"
}
patient = Patient(**patient_data)
assert patient.resourceType == "Patient"
assert patient.id == "test-patient"
assert patient.active is True
assert len(patient.name) == 1
assert patient.name[0].family == "Doe"
assert patient.gender == "male"
def test_invalid_patient_gender(self):
"""Test patient with invalid gender value"""
patient_data = {
"resourceType": "Patient",
"gender": "invalid-gender"
}
with pytest.raises(ValidationError) as exc_info:
Patient(**patient_data)
assert "gender" in str(exc_info.value)
def test_patient_serialization(self):
"""Test patient serialization to JSON"""
patient = Patient(
resourceType="Patient",
id="test-patient",
active=True,
name=[{"family": "Doe", "given": ["John"]}],
gender="male"
)
json_data = patient.json(exclude_none=True)
parsed_data = json.loads(json_data)
assert parsed_data["resourceType"] == "Patient"
assert parsed_data["id"] == "test-patient"
assert parsed_data["active"] is True
def test_observation_with_quantity_value(self):
"""Test observation with quantity value"""
obs_data = {
"resourceType": "Observation",
"status": "final",
"code": {
"coding": [{
"system": "http://loinc.org",
"code": "55284-4",
"display": "Blood pressure"
}]
},
"subject": {"reference": "Patient/123"},
"valueQuantity": {
"value": 120,
"unit": "mmHg",
"system": "http://unitsofmeasure.org",
"code": "mm[Hg]"
}
}
observation = Observation(**obs_data)
assert observation.status == "final"
assert observation.valueQuantity.value == 120
assert observation.valueQuantity.unit == "mmHg"
def test_observation_missing_required_fields(self):
"""Test observation missing required fields"""
obs_data = {
"resourceType": "Observation"
# Missing status, code, subject
}
with pytest.raises(ValidationError) as exc_info:
Observation(**obs_data)
error_msg = str(exc_info.value)
assert "status" in error_msg
assert "code" in error_msg
@pytest.fixture
def sample_patient():
"""Fixture providing a sample patient"""
return Patient(
resourceType="Patient",
id="sample-patient",
active=True,
name=[{
"family": "Smith",
"given": ["Jane"]
}],
gender="female",
birthDate="1985-03-15"
)
def test_patient_fixture(sample_patient):
"""Test using patient fixture"""
assert sample_patient.name[0].family == "Smith"
assert sample_patient.gender == "female"
```
## API Integration Tests
### FHIR Server Testing
```python
import pytest
import httpx
import asyncio
from typing import AsyncGenerator
@pytest.fixture
async def test_client() -> AsyncGenerator[httpx.AsyncClient, None]:
"""Async test client for FHIR server"""
async with httpx.AsyncClient(
base_url="http://localhost:8000",
timeout=30.0
) as client:
yield client
@pytest.fixture
def patient_data():
"""Sample patient data for testing"""
return {
"resourceType": "Patient",
"active": True,
"name": [{
"family": "TestPatient",
"given": ["Integration"]
}],
"gender": "unknown",
"birthDate": "1990-01-01"
}
class TestFHIRServerAPI:
@pytest.mark.asyncio
async def test_server_metadata(self, test_client):
"""Test capability statement endpoint"""
response = await test_client.get("/metadata")
assert response.status_code == 200
data = response.json()
assert data["resourceType"] == "CapabilityStatement"
assert data["fhirVersion"] == "4.0.1"
assert "rest" in data
# Check supported resources
resources = data["rest"][0]["resource"]
resource_types = [r["type"] for r in resources]
assert "Patient" in resource_types
assert "Observation" in resource_types
@pytest.mark.asyncio
async def test_patient_crud_operations(self, test_client, patient_data):
"""Test complete patient CRUD cycle"""
# CREATE
create_response = await test_client.post(
"/Patient",
json=patient_data,
headers={"Content-Type": "application/fhir+json"}
)
assert create_response.status_code == 201
created_patient = create_response.json()
patient_id = created_patient["id"]
assert created_patient["resourceType"] == "Patient"
assert "meta" in created_patient
# READ
read_response = await test_client.get(f"/Patient/{patient_id}")
assert read_response.status_code == 200
read_patient = read_response.json()
assert read_patient["id"] == patient_id
assert read_patient["name"][0]["family"] == "TestPatient"
# UPDATE
read_patient["active"] = False
update_response = await test_client.put(
f"/Patient/{patient_id}",
json=read_patient,
headers={"Content-Type": "application/fhir+json"}
)
assert update_response.status_code == 200
updated_patient = update_response.json()
assert updated_patient["active"] is False
assert int(updated_patient["meta"]["versionId"]) > int(created_patient["meta"]["versionId"])
# DELETE
delete_response = await test_client.delete(f"/Patient/{patient_id}")
assert delete_response.status_code == 204
# Verify deletion
get_deleted_response = await test_client.get(f"/Patient/{patient_id}")
assert get_deleted_response.status_code == 404
@pytest.mark.asyncio
async def test_patient_search(self, test_client):
"""Test patient search functionality"""
# Create test patients
patients = [
{
"resourceType": "Patient",
"name": [{"family": "Smith", "given": ["John"]}],
"gender": "male",
"birthDate": "1980-01-01"
},
{
"resourceType": "Patient",
"name": [{"family": "Smith", "given": ["Jane"]}],
"gender": "female",
"birthDate": "1985-05-15"
},
{
"resourceType": "Patient",
"name": [{"family": "Doe", "given": ["Bob"]}],
"gender": "male",
"birthDate": "1990-12-25"
}
]
created_ids = []
for patient_data in patients:
response = await test_client.post(
"/Patient",
json=patient_data,
headers={"Content-Type": "application/fhir+json"}
)
assert response.status_code == 201
created_ids.append(response.json()["id"])
try:
# Search by family name
search_response = await test_client.get("/Patient?family=Smith")
assert search_response.status_code == 200
search_results = search_response.json()
assert search_results["resourceType"] == "Bundle"
assert search_results["total"] == 2
# Search by gender
search_response = await test_client.get("/Patient?gender=male")
assert search_response.status_code == 200
search_results = search_response.json()
assert search_results["total"] == 2
# Search by birth date
search_response = await test_client.get("/Patient?birthdate=1980-01-01")
assert search_response.status_code == 200
search_results = search_response.json()
assert search_results["total"] == 1
assert search_results["entry"][0]["resource"]["name"][0]["given"][0] == "John"
finally:
# Cleanup
for patient_id in created_ids:
await test_client.delete(f"/Patient/{patient_id}")
@pytest.mark.asyncio
async def test_invalid_resource_handling(self, test_client):
"""Test error handling for invalid resources"""
# Invalid resource type
invalid_data = {
"resourceType": "InvalidResource",
"someField": "someValue"
}
response = await test_client.post(
"/Patient",
json=invalid_data,
headers={"Content-Type": "application/fhir+json"}
)
assert response.status_code == 400
error_response = response.json()
assert "OperationOutcome" in str(error_response) or "detail" in error_response
@pytest.mark.asyncio
async def test_content_type_validation(self, test_client, patient_data):
"""Test content type validation"""
# Missing content type
response = await test_client.post("/Patient", json=patient_data)
# Should accept application/json as fallback
assert response.status_code in [201, 415] # Depends on implementation
# Correct FHIR content type
response = await test_client.post(
"/Patient",
json=patient_data,
headers={"Content-Type": "application/fhir+json"}
)
assert response.status_code == 201
```
## Validation Testing
### Profile Validation Tests
```python
import pytest
from unittest.mock import Mock, patch
from fhir_validator import FHIRValidator, StructureDefinitionLoader
class TestFHIRValidation:
@pytest.fixture
def validator(self):
"""Create validator with mock package manager"""
mock_package_manager = Mock()
return FHIRValidator(mock_package_manager)
@pytest.fixture
def us_core_patient_profile(self):
"""Mock US Core Patient profile"""
return {
"resourceType": "StructureDefinition",
"url": "http://hl7.org/fhir/us/core/StructureDefinition/us-core-patient",
"differential": {
"element": [
{
"path": "Patient.identifier",
"min": 1,
"mustSupport": True
},
{
"path": "Patient.name",
"min": 1,
"mustSupport": True
}
]
}
}
def test_valid_us_core_patient(self, validator):
"""Test validation against US Core Patient profile"""
patient_data = {
"resourceType": "Patient",
"id": "example",
"meta": {
"profile": ["http://hl7.org/fhir/us/core/StructureDefinition/us-core-patient"]
},
"identifier": [{
"system": "http://hospital.org/patients",
"value": "123456"
}],
"name": [{
"family": "Doe",
"given": ["John"]
}],
"gender": "male"
}
with patch.object(
validator.structure_loader,
'load_structure_definition',
return_value=us_core_patient_profile()
):
result = validator.validate_resource(
patient_data,
["http://hl7.org/fhir/us/core/StructureDefinition/us-core-patient"]
)
assert result.valid
assert len(result.errors) == 0
def test_invalid_us_core_patient_missing_identifier(self, validator, us_core_patient_profile):
"""Test US Core Patient missing required identifier"""
patient_data = {
"resourceType": "Patient",
"name": [{
"family": "Doe",
"given": ["John"]
}]
}
with patch.object(
validator.structure_loader,
'load_structure_definition',
return_value=us_core_patient_profile
):
result = validator.validate_resource(
patient_data,
["http://hl7.org/fhir/us/core/StructureDefinition/us-core-patient"]
)
assert not result.valid
assert any("identifier" in error.lower() for error in result.errors)
def test_terminology_validation(self, validator):
"""Test validation of coded values against ValueSets"""
observation_data = {
"resourceType": "Observation",
"status": "final",
"code": {
"coding": [{
"system": "http://loinc.org",
"code": "55284-4",
"display": "Blood pressure"
}]
},
"subject": {"reference": "Patient/123"},
"valueQuantity": {
"value": 120,
"unit": "mmHg",
"system": "http://unitsofmeasure.org",
"code": "mm[Hg]"
}
}
# Mock terminology validation
with patch.object(
validator.terminology_validator,
'validate_coding',
return_value=[]
):
result = validator.validate_resource(observation_data)
assert result.valid
```
## Performance Testing
### Load Testing with Locust
```python
from locust import HttpUser, task, between
import json
import random
class FHIRServerUser(HttpUser):
wait_time = between(1, 3)
def on_start(self):
"""Setup test data"""
self.patient_ids = []
self.observation_ids = []
# Create some test patients
for i in range(5):
patient_data = {
"resourceType": "Patient",
"name": [{
"family": f"TestFamily{i}",
"given": [f"TestGiven{i}"]
}],
"gender": random.choice(["male", "female"]),
"birthDate": f"198{i}-01-01"
}
response = self.client.post(
"/Patient",
json=patient_data,
headers={"Content-Type": "application/fhir+json"}
)
if response.status_code == 201:
self.patient_ids.append(response.json()["id"])
@task(3)
def search_patients(self):
"""Search for patients (most common operation)"""
search_params = [
"?family=Test",
"?gender=male",
"?birthdate=ge1980",
"?_count=10"
]
param = random.choice(search_params)
self.client.get(f"/Patient{param}")
@task(2)
def read_patient(self):
"""Read specific patient"""
if self.patient_ids:
patient_id = random.choice(self.patient_ids)
self.client.get(f"/Patient/{patient_id}")
@task(1)
def create_observation(self):
"""Create observation for existing patient"""
if self.patient_ids:
patient_id = random.choice(self.patient_ids)
observation_data = {
"resourceType": "Observation",
"status": "final",
"code": {
"coding": [{
"system": "http://loinc.org",
"code": "55284-4",
"display": "Blood pressure"
}]
},
"subject": {"reference": f"Patient/{patient_id}"},
"valueQuantity": {
"value": random.randint(90, 180),
"unit": "mmHg",
"system": "http://unitsofmeasure.org",
"code": "mm[Hg]"
}
}
response = self.client.post(
"/Observation",
json=observation_data,
headers={"Content-Type": "application/fhir+json"}
)
if response.status_code == 201:
self.observation_ids.append(response.json()["id"])
@task(1)
def get_server_metadata(self):
"""Get server capability statement"""
self.client.get("/metadata")
```
## Conformance Testing
### FHIR Test Suite Integration
```python
import pytest
import json
import requests
from pathlib import Path
class TestFHIRConformance:
"""Test suite for FHIR conformance testing"""
@pytest.fixture(scope="class")
def test_server_url(self):
"""FHIR server URL for testing"""
return "http://localhost:8000"
@pytest.fixture(scope="class")
def capability_statement(self, test_server_url):
"""Get server capability statement"""
response = requests.get(f"{test_server_url}/metadata")
assert response.status_code == 200
return response.json()
def test_capability_statement_structure(self, capability_statement):
"""Test capability statement has required fields"""
assert capability_statement["resourceType"] == "CapabilityStatement"
assert "fhirVersion" in capability_statement
assert "rest" in capability_statement
assert len(capability_statement["rest"]) > 0
rest = capability_statement["rest"][0]
assert rest["mode"] == "server"
assert "resource" in rest
def test_supported_resources(self, capability_statement):
"""Test server supports required resource types"""
rest = capability_statement["rest"][0]
supported_types = [r["type"] for r in rest["resource"]]
# Check minimum required resources
required_resources = ["Patient", "Observation"]
for resource_type in required_resources:
assert resource_type in supported_types
def test_search_parameters(self, capability_statement):
"""Test search parameters are properly defined"""
rest = capability_statement["rest"][0]
for resource in rest["resource"]:
if resource["type"] == "Patient":
search_params = {p["name"]: p["type"] for p in resource.get("searchParam", [])}
# Check required Patient search parameters
assert "name" in search_params
assert "family" in search_params
assert "birthdate" in search_params
# Verify parameter types
assert search_params["name"] == "string"
assert search_params["birthdate"] == "date"
@pytest.mark.parametrize("format_type", [
"application/fhir+json",
"application/json",
"application/fhir+xml" # If supported
])
def test_supported_formats(self, test_server_url, capability_statement, format_type):
"""Test server supports advertised formats"""
supported_formats = capability_statement.get("format", [])
if format_type in supported_formats:
headers = {"Accept": format_type}
response = requests.get(f"{test_server_url}/metadata", headers=headers)
assert response.status_code == 200
if "json" in format_type:
# Should be valid JSON
response.json()
elif "xml" in format_type:
# Should be valid XML (basic check)
assert response.text.startswith("<?xml") or response.text.startswith("<")
def test_fhir_examples_validation():
"""Test validation of official FHIR examples"""
examples_dir = Path("test_data/fhir_examples")
if not examples_dir.exists():
pytest.skip("FHIR examples directory not found")
validator = FHIRValidator(None) # Mock package manager
for example_file in examples_dir.glob("*.json"):
with open(example_file) as f:
resource_data = json.load(f)
# Basic validation
result = validator.validate_resource(resource_data)
if not result.valid:
pytest.fail(
f"Official example {example_file.name} failed validation: "
f"{', '.join(result.errors)}"
)
class TestInteroperability:
"""Test interoperability with other FHIR servers"""
@pytest.mark.parametrize("server_url", [
"https://hapi.fhir.org/baseR4",
"https://launch.smarthealthit.org/v/r4/fhir",
# Add other test servers
])
def test_cross_server_patient_read(self, server_url):
"""Test reading patients from different FHIR servers"""
try:
# Try to read a known test patient
response = requests.get(
f"{server_url}/Patient",
headers={"Accept": "application/fhir+json"},
timeout=10
)
if response.status_code == 200:
bundle = response.json()
assert bundle["resourceType"] == "Bundle"
# Validate first patient if available
if bundle.get("entry"):
patient = bundle["entry"][0]["resource"]
assert patient["resourceType"] == "Patient"
except requests.RequestException:
pytest.skip(f"Server {server_url} not accessible")
```
## Test Data Management
### Test Data Factory
```python
import factory
from faker import Faker
from datetime import datetime, timedelta
import random
fake = Faker()
class PatientFactory(factory.Factory):
"""Factory for generating test Patient resources"""
class Meta:
model = dict
resourceType = "Patient"
id = factory.LazyAttribute(lambda obj: fake.uuid4())
active = True
@factory.LazyAttribute
def name(obj):
return [{
"family": fake.last_name(),
"given": [fake.first_name(), fake.first_name()]
}]
gender = factory.LazyAttribute(
lambda obj: random.choice(["male", "female", "other", "unknown"])
)
@factory.LazyAttribute
def birthDate(obj):
birth_date = fake.date_between(start_date="-80y", end_date="-18y")
return birth_date.isoformat()
@factory.LazyAttribute
def identifier(obj):
return [{
"system": "http://hospital.org/patients",
"value": fake.random_number(digits=8)
}]
class ObservationFactory(factory.Factory):
"""Factory for generating test Observation resources"""
class Meta:
model = dict
resourceType = "Observation"
id = factory.LazyAttribute(lambda obj: fake.uuid4())
status = "final"
@factory.LazyAttribute
def code(obj):
return {
"coding": [{
"system": "http://loinc.org",
"code": random.choice(["55284-4", "8480-6", "8462-4"]),
"display": random.choice(["Blood pressure", "Systolic BP", "Diastolic BP"])
}]
}
@factory.LazyAttribute
def subject(obj):
return {"reference": f"Patient/{fake.uuid4()}"}
@factory.LazyAttribute
def effectiveDateTime(obj):
effect_time = fake.date_time_between(start_date="-1y", end_date="now")
return effect_time.isoformat()
@factory.LazyAttribute
def valueQuantity(obj):
return {
"value": random.randint(80, 180),
"unit": "mmHg",
"system": "http://unitsofmeasure.org",
"code": "mm[Hg]"
}
# Usage examples
def test_with_factory_data():
"""Test using factory-generated data"""
patient = PatientFactory()
observation = ObservationFactory(subject={"reference": f"Patient/{patient['id']}"})
assert patient["resourceType"] == "Patient"
assert observation["subject"]["reference"].startswith("Patient/")
# Create multiple resources
patients = PatientFactory.create_batch(10)
assert len(patients) == 10
```
## Continuous Integration
### GitHub Actions FHIR Testing
```yaml
# .github/workflows/fhir-tests.yml
name: FHIR Tests
on:
push:
branches: [ main, develop ]
pull_request:
branches: [ main ]
jobs:
test:
runs-on: ubuntu-latest
strategy:
matrix:
python-version: [3.9, 3.10, 3.11]
services:
postgres:
image: postgres:13
env:
POSTGRES_PASSWORD: postgres
POSTGRES_DB: fhir_test
options: >-
--health-cmd pg_isready
--health-interval 10s
--health-timeout 5s
--health-retries 5
ports:
- 5432:5432
steps:
- uses: actions/checkout@v3
- name: Set up Python ${{ matrix.python-version }}
uses: actions/setup-python@v4
with:
python-version: ${{ matrix.python-version }}
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install -r requirements.txt
pip install -r requirements-test.txt
- name: Install FHIR packages
run: |
python scripts/fhir_package_manager.py install hl7.fhir.r4.core 4.0.1
python scripts/fhir_package_manager.py install hl7.fhir.us.core 5.0.1
- name: Run FHIR validation tests
run: |
python scripts/fhir_validator.py validate test_data/patient_example.json
python scripts/fhir_validator.py validate test_data/observation_example.json
- name: Run unit tests
run: |
pytest tests/unit/ -v --cov=src
- name: Start FHIR server
run: |
python src/fhir_server.py &
sleep 5
env:
DATABASE_URL: postgresql://postgres:postgres@localhost:5432/fhir_test
- name: Run integration tests
run: |
pytest tests/integration/ -v
- name: Run conformance tests
run: |
pytest tests/conformance/ -v
- name: Upload coverage reports
uses: codecov/codecov-action@v3
```
This comprehensive testing reference provides patterns for ensuring FHIR implementation quality across all testing levels, from unit tests to full conformance testing.
references/validation_patterns.md# FHIR Validation and StructureDefinition Reference
This document provides comprehensive guidance on FHIR resource validation, profile validation, and working with StructureDefinitions.
## Validation Levels
### 1. Structure Validation
Validates basic FHIR structure and data types.
```python
def validate_fhir_structure(resource: dict) -> ValidationResult:
"""Validate basic FHIR resource structure"""
errors = []
warnings = []
# Required fields
if not resource.get('resourceType'):
errors.append("Missing required field: resourceType")
# ID validation
if 'id' in resource:
if not re.match(r'^[A-Za-z0-9\-\.]{1,64}$', resource['id']):
errors.append("Invalid id format")
# Meta validation
if 'meta' in resource:
meta = resource['meta']
if 'versionId' in meta and not re.match(r'^[A-Za-z0-9\-\.]{1,64}$', meta['versionId']):
errors.append("Invalid meta.versionId format")
return ValidationResult(len(errors) == 0, errors, warnings)
```
### 2. Data Type Validation
Validates FHIR primitive and complex data types.
```python
from datetime import datetime
import re
def validate_fhir_datatypes(element: any, element_type: str) -> list:
"""Validate FHIR data types"""
errors = []
if element_type == 'date':
if not re.match(r'^\d{4}(-\d{2}(-\d{2})?)?$', str(element)):
errors.append(f"Invalid date format: {element}")
elif element_type == 'dateTime':
try:
# FHIR dateTime format: YYYY-MM-DDTHH:mm:ss+zz:zz
datetime.fromisoformat(str(element).replace('Z', '+00:00'))
except ValueError:
errors.append(f"Invalid dateTime format: {element}")
elif element_type == 'boolean':
if element not in [True, False]:
errors.append(f"Invalid boolean value: {element}")
elif element_type == 'integer':
if not isinstance(element, int) or element < -2147483648 or element > 2147483647:
errors.append(f"Invalid integer value: {element}")
elif element_type == 'decimal':
if not isinstance(element, (int, float)):
errors.append(f"Invalid decimal value: {element}")
elif element_type == 'uri':
# Basic URI validation
if not re.match(r'^[a-zA-Z][a-zA-Z0-9+.-]*:', str(element)):
errors.append(f"Invalid URI format: {element}")
elif element_type == 'url':
# URL should be absolute
if not re.match(r'^https?://', str(element)):
errors.append(f"Invalid URL format: {element}")
elif element_type == 'canonical':
# Can be URL or URI
if not (re.match(r'^https?://', str(element)) or
re.match(r'^[a-zA-Z][a-zA-Z0-9+.-]*:', str(element))):
errors.append(f"Invalid canonical format: {element}")
return errors
```
### 3. Cardinality Validation
Validates min/max occurrences of elements.
```python
def validate_cardinality(resource: dict, structure_def: dict) -> list:
"""Validate element cardinality against StructureDefinition"""
errors = []
for element in structure_def.get('differential', {}).get('element', []):
path = element['path']
min_occurs = element.get('min', 0)
max_occurs = element.get('max', '*')
# Extract value from resource using path
values = extract_values_by_path(resource, path)
value_count = len(values)
# Check minimum cardinality
if value_count < min_occurs:
errors.append(f"{path}: minimum {min_occurs} occurrences required, found {value_count}")
# Check maximum cardinality
if max_occurs != '*' and value_count > int(max_occurs):
errors.append(f"{path}: maximum {max_occurs} occurrences allowed, found {value_count}")
return errors
def extract_values_by_path(resource: dict, fhir_path: str) -> list:
"""Extract values from resource using FHIRPath expression"""
# Simplified implementation - real implementation would use FHIRPath evaluator
parts = fhir_path.split('.')
current = resource
try:
for part in parts[1:]: # Skip resource type
if '[' in part:
# Handle array access like 'name[0]'
field, index = part.split('[')
index = int(index.rstrip(']'))
current = current[field][index]
else:
current = current[part]
return [current] if current is not None else []
except (KeyError, IndexError, TypeError):
return []
```
## Profile Validation
### Loading StructureDefinitions
```python
class StructureDefinitionLoader:
def __init__(self, package_manager):
self.package_manager = package_manager
self.cache = {}
def load_structure_definition(self, url: str) -> dict:
"""Load StructureDefinition by canonical URL"""
if url in self.cache:
return self.cache[url]
# Search in loaded packages
for package_id, version in self.get_loaded_packages():
structure_defs = self.package_manager.search_resources(
package_id, version, resource_type='StructureDefinition'
)
for sd in structure_defs:
if sd.get('url') == url:
full_sd = self.load_resource_file(sd['file_path'])
self.cache[url] = full_sd
return full_sd
raise ValueError(f"StructureDefinition not found: {url}")
def get_loaded_packages(self) -> list:
"""Get list of loaded package IDs and versions"""
return [
('hl7.fhir.r4.core', '4.0.1'),
('hl7.fhir.us.core', '5.0.1'),
# Add other loaded packages
]
def load_resource_file(self, file_path: str) -> dict:
"""Load FHIR resource from file"""
with open(file_path, 'r') as f:
return json.load(f)
```
### Profile-based Validation
```python
def validate_against_profile(resource: dict, profile_url: str,
loader: StructureDefinitionLoader) -> ValidationResult:
"""Validate resource against specific profile"""
errors = []
warnings = []
try:
profile = loader.load_structure_definition(profile_url)
# Validate must-support elements
must_support_errors = validate_must_support(resource, profile)
errors.extend(must_support_errors)
# Validate constraints
constraint_errors = validate_constraints(resource, profile)
errors.extend(constraint_errors)
# Validate cardinality
cardinality_errors = validate_cardinality(resource, profile)
errors.extend(cardinality_errors)
# Validate fixed values
fixed_value_errors = validate_fixed_values(resource, profile)
errors.extend(fixed_value_errors)
# Validate binding strengths
binding_warnings = validate_value_set_bindings(resource, profile)
warnings.extend(binding_warnings)
except Exception as e:
errors.append(f"Profile validation error: {str(e)}")
return ValidationResult(len(errors) == 0, errors, warnings)
def validate_must_support(resource: dict, profile: dict) -> list:
"""Validate mustSupport elements are present"""
errors = []
for element in profile.get('differential', {}).get('element', []):
if element.get('mustSupport'):
path = element['path']
values = extract_values_by_path(resource, path)
if not values:
errors.append(f"Missing mustSupport element: {path}")
return errors
def validate_constraints(resource: dict, profile: dict) -> list:
"""Validate FHIRPath constraints"""
errors = []
for element in profile.get('differential', {}).get('element', []):
for constraint in element.get('constraint', []):
expression = constraint.get('expression')
severity = constraint.get('severity', 'error')
if expression:
# Evaluate FHIRPath expression
try:
result = evaluate_fhirpath(resource, expression)
if not result:
message = constraint.get('human', f"Constraint violation: {expression}")
if severity == 'error':
errors.append(f"{element['path']}: {message}")
# warnings would be handled similarly
except Exception as e:
errors.append(f"Error evaluating constraint {constraint.get('key', '')}: {str(e)}")
return errors
```
## ValueSet and CodeSystem Validation
### Terminology Validation
```python
class TerminologyValidator:
def __init__(self, package_manager):
self.package_manager = package_manager
self.value_sets = {}
self.code_systems = {}
def validate_coding(self, coding: dict, binding: dict) -> list:
"""Validate Coding against ValueSet binding"""
errors = []
warnings = []
if not coding:
return errors
value_set_url = binding.get('valueSet')
strength = binding.get('strength', 'required')
if not value_set_url:
return errors
try:
value_set = self.load_value_set(value_set_url)
is_valid = self.coding_in_value_set(coding, value_set)
if not is_valid:
message = f"Invalid code {coding.get('code')} from system {coding.get('system')}"
if strength == 'required':
errors.append(message)
elif strength in ['extensible', 'preferred']:
warnings.append(f"Recommended: {message}")
# 'example' strength doesn't generate errors/warnings
except Exception as e:
errors.append(f"Terminology validation error: {str(e)}")
return errors
def load_value_set(self, url: str) -> dict:
"""Load ValueSet by URL"""
if url in self.value_sets:
return self.value_sets[url]
# Search in packages
for package_id, version in self.get_loaded_packages():
value_sets = self.package_manager.search_resources(
package_id, version, resource_type='ValueSet'
)
for vs in value_sets:
if vs.get('url') == url:
full_vs = self.load_resource_file(vs['file_path'])
self.value_sets[url] = full_vs
return full_vs
raise ValueError(f"ValueSet not found: {url}")
def coding_in_value_set(self, coding: dict, value_set: dict) -> bool:
"""Check if Coding is in ValueSet"""
system = coding.get('system')
code = coding.get('code')
if not system or not code:
return False
# Check expansion if present
expansion = value_set.get('expansion')
if expansion:
for concept in expansion.get('contains', []):
if (concept.get('system') == system and
concept.get('code') == code):
return True
# Check compose rules
compose = value_set.get('compose')
if compose:
for include in compose.get('include', []):
if self.coding_matches_include(coding, include):
return True
return False
def coding_matches_include(self, coding: dict, include: dict) -> bool:
"""Check if coding matches ValueSet include rules"""
system = coding.get('system')
code = coding.get('code')
include_system = include.get('system')
if include_system and include_system != system:
return False
# Check specific concepts
concepts = include.get('concept', [])
if concepts:
return any(c.get('code') == code for c in concepts)
# Check filters
filters = include.get('filter', [])
for filter_rule in filters:
if not self.coding_matches_filter(coding, filter_rule):
return False
# If no specific concepts or filters, includes all codes from system
return include_system == system
```
## Slicing Validation
### Slice Discrimination
```python
def validate_slicing(resource: dict, element_def: dict) -> list:
"""Validate array slicing rules"""
errors = []
path = element_def['path']
slicing = element_def.get('slicing')
if not slicing:
return errors
# Extract array values
values = extract_values_by_path(resource, path)
# Group values by discriminator
discriminator = slicing['discriminator'][0] # Simplified - handle multiple discriminators
disc_type = discriminator['type']
disc_path = discriminator['path']
groups = {}
for value in values:
disc_value = extract_discriminator_value(value, disc_path, disc_type)
if disc_value not in groups:
groups[disc_value] = []
groups[disc_value].append(value)
# Validate slice cardinality
slice_elements = get_slice_elements(element_def['path'])
for slice_element in slice_elements:
slice_name = get_slice_name(slice_element)
expected_disc_value = get_slice_discriminator_value(slice_element)
actual_count = len(groups.get(expected_disc_value, []))
min_occurs = slice_element.get('min', 0)
max_occurs = slice_element.get('max', '*')
if actual_count < min_occurs:
errors.append(f"Slice {slice_name}: minimum {min_occurs} required, found {actual_count}")
if max_occurs != '*' and actual_count > int(max_occurs):
errors.append(f"Slice {slice_name}: maximum {max_occurs} allowed, found {actual_count}")
return errors
def extract_discriminator_value(element: dict, path: str, disc_type: str) -> str:
"""Extract discriminator value for slicing"""
if disc_type == 'value':
# Extract value at specified path
return str(extract_values_by_path(element, path)[0] if
extract_values_by_path(element, path) else '')
elif disc_type == 'type':
# Discriminate by element type
return element.get('resourceType', type(element).__name__)
elif disc_type == 'profile':
# Discriminate by profile in meta
profiles = element.get('meta', {}).get('profile', [])
return profiles[0] if profiles else ''
elif disc_type == 'pattern':
# Complex pattern matching would go here
pass
return ''
```
## Extension Validation
### Extension Handling
```python
def validate_extensions(resource: dict, structure_def: dict) -> list:
"""Validate extensions in resource"""
errors = []
# Collect all extensions
extensions = collect_extensions(resource)
for extension in extensions:
url = extension.get('url')
if not url:
errors.append("Extension missing required 'url' element")
continue
try:
# Load extension definition
ext_def = load_extension_definition(url)
# Validate extension structure
ext_errors = validate_extension_structure(extension, ext_def)
errors.extend(ext_errors)
except Exception as e:
errors.append(f"Error validating extension {url}: {str(e)}")
return errors
def collect_extensions(resource: dict, path: str = '') -> list:
"""Recursively collect all extensions from resource"""
extensions = []
if isinstance(resource, dict):
# Direct extensions
if 'extension' in resource:
for ext in resource['extension']:
extensions.append(ext)
# Primitive extensions (e.g., _birthDate)
for key, value in resource.items():
if key.startswith('_') and isinstance(value, dict):
if 'extension' in value:
extensions.extend(value['extension'])
# Recurse into nested objects
for key, value in resource.items():
if key not in ['extension'] and not key.startswith('_'):
if isinstance(value, (dict, list)):
nested_extensions = collect_extensions(value, f"{path}.{key}")
extensions.extend(nested_extensions)
elif isinstance(resource, list):
for i, item in enumerate(resource):
if isinstance(item, dict):
nested_extensions = collect_extensions(item, f"{path}[{i}]")
extensions.extend(nested_extensions)
return extensions
```
## Validation Orchestration
### Complete Resource Validation
```python
class FHIRValidator:
def __init__(self, package_manager):
self.package_manager = package_manager
self.structure_loader = StructureDefinitionLoader(package_manager)
self.terminology_validator = TerminologyValidator(package_manager)
def validate_resource(self, resource: dict,
profile_urls: list = None) -> ValidationResult:
"""Complete FHIR resource validation"""
all_errors = []
all_warnings = []
# 1. Structure validation
structure_result = validate_fhir_structure(resource)
all_errors.extend(structure_result.errors)
all_warnings.extend(structure_result.warnings)
# 2. Data type validation
datatype_errors = self.validate_resource_datatypes(resource)
all_errors.extend(datatype_errors)
# 3. Profile validation
if profile_urls:
for profile_url in profile_urls:
profile_result = validate_against_profile(
resource, profile_url, self.structure_loader
)
all_errors.extend(profile_result.errors)
all_warnings.extend(profile_result.warnings)
else:
# Validate against base resource definition
base_profile_url = f"http://hl7.org/fhir/StructureDefinition/{resource['resourceType']}"
try:
profile_result = validate_against_profile(
resource, base_profile_url, self.structure_loader
)
all_errors.extend(profile_result.errors)
all_warnings.extend(profile_result.warnings)
except Exception as e:
all_warnings.append(f"Could not validate against base profile: {str(e)}")
return ValidationResult(
valid=(len(all_errors) == 0),
errors=all_errors,
warnings=all_warnings
)
def validate_bundle(self, bundle: dict) -> ValidationResult:
"""Validate FHIR Bundle and all contained resources"""
all_errors = []
all_warnings = []
# Validate bundle structure
bundle_result = self.validate_resource(bundle)
all_errors.extend(bundle_result.errors)
all_warnings.extend(bundle_result.warnings)
# Validate each entry
for i, entry in enumerate(bundle.get('entry', [])):
if 'resource' in entry:
resource = entry['resource']
# Extract profile URLs from meta
profile_urls = resource.get('meta', {}).get('profile', [])
resource_result = self.validate_resource(resource, profile_urls)
# Prefix errors with entry index
for error in resource_result.errors:
all_errors.append(f"Bundle.entry[{i}]: {error}")
for warning in resource_result.warnings:
all_warnings.append(f"Bundle.entry[{i}]: {warning}")
return ValidationResult(
valid=(len(all_errors) == 0),
errors=all_errors,
warnings=all_warnings
)
```
## Performance Optimization
### Validation Caching
```python
from functools import lru_cache
import hashlib
class CachedValidator:
def __init__(self, base_validator):
self.base_validator = base_validator
self.resource_cache = {}
@lru_cache(maxsize=1000)
def get_structure_definition(self, url: str):
"""Cached StructureDefinition loading"""
return self.base_validator.structure_loader.load_structure_definition(url)
def validate_resource_cached(self, resource: dict, profile_urls: list = None):
"""Validate with caching for identical resources"""
# Create hash of resource for caching
resource_str = json.dumps(resource, sort_keys=True)
cache_key = hashlib.md5(resource_str.encode()).hexdigest()
if profile_urls:
cache_key += '_' + '_'.join(sorted(profile_urls))
if cache_key in self.resource_cache:
return self.resource_cache[cache_key]
result = self.base_validator.validate_resource(resource, profile_urls)
self.resource_cache[cache_key] = result
return result
```
This reference provides comprehensive patterns for implementing robust FHIR validation in your applications, covering all major aspects from basic structure validation to complex profile and terminology validation.
scripts/fhir_package_manager.py#!/usr/bin/env python3
"""
FHIR Package Manager - Download, cache, and manage FHIR packages locally
"""
import os
import json
import requests
import tarfile
import zipfile
from pathlib import Path
from typing import Dict, List, Optional, Any
import argparse
import sys
class FHIRPackageManager:
def __init__(self, cache_dir: str = None):
self.cache_dir = Path(cache_dir or Path.home() / ".fhir" / "packages")
self.cache_dir.mkdir(parents=True, exist_ok=True)
self.registry_url = "https://packages.fhir.org"
def install_package(self, package_id: str, version: str = "latest") -> Path:
"""Download and install a FHIR package"""
print(f"Installing {package_id}@{version}")
# Check if already cached
package_dir = self.cache_dir / package_id / version
if package_dir.exists():
print(f"Package already cached at {package_dir}")
return package_dir
# Download package metadata
metadata = self._get_package_metadata(package_id, version)
if not metadata:
raise ValueError(f"Package {package_id}@{version} not found")
# Download package
download_url = metadata.get("dist", {}).get("tarball")
if not download_url:
raise ValueError(f"No download URL found for {package_id}@{version}")
package_dir.mkdir(parents=True, exist_ok=True)
self._download_and_extract(download_url, package_dir)
print(f"Package installed to {package_dir}")
return package_dir
def _get_package_metadata(self, package_id: str, version: str) -> Optional[Dict]:
"""Fetch package metadata from FHIR registry"""
try:
url = f"{self.registry_url}/{package_id}"
if version != "latest":
url += f"/{version}"
response = requests.get(url)
response.raise_for_status()
return response.json()
except requests.RequestException as e:
print(f"Error fetching metadata: {e}")
return None
def _download_and_extract(self, url: str, destination: Path):
"""Download and extract package archive"""
response = requests.get(url, stream=True)
response.raise_for_status()
# Determine archive type and extract
if url.endswith('.tgz') or url.endswith('.tar.gz'):
with tarfile.open(fileobj=response.raw, mode='r:gz') as tar:
tar.extractall(destination)
elif url.endswith('.zip'):
import io
with zipfile.ZipFile(io.BytesIO(response.content)) as zip_file:
zip_file.extractall(destination)
else:
raise ValueError(f"Unsupported archive format: {url}")
def list_installed(self) -> List[Dict[str, str]]:
"""List installed packages"""
packages = []
if not self.cache_dir.exists():
return packages
for package_dir in self.cache_dir.iterdir():
if package_dir.is_dir():
for version_dir in package_dir.iterdir():
if version_dir.is_dir():
packages.append({
"id": package_dir.name,
"version": version_dir.name,
"path": str(version_dir)
})
return packages
def load_package_manifest(self, package_id: str, version: str = "latest") -> Optional[Dict]:
"""Load package.json manifest from installed package"""
package_dir = self.cache_dir / package_id / version
manifest_path = package_dir / "package.json"
if not manifest_path.exists():
return None
with open(manifest_path, 'r') as f:
return json.load(f)
def get_resource_files(self, package_id: str, version: str = "latest",
resource_type: str = None) -> List[Path]:
"""Get list of FHIR resource files from package"""
package_dir = self.cache_dir / package_id / version
if not package_dir.exists():
return []
# Look for resources in common locations
search_dirs = [
package_dir / "package",
package_dir,
package_dir / "input"
]
resource_files = []
for search_dir in search_dirs:
if search_dir.exists():
pattern = "*.json"
if resource_type:
pattern = f"*{resource_type}*.json"
resource_files.extend(search_dir.rglob(pattern))
return resource_files
def build_resource_index(self, package_id: str, version: str = "latest") -> Dict[str, List[Dict]]:
"""Build searchable index of FHIR resources in package"""
index = {
"StructureDefinition": [],
"ValueSet": [],
"CodeSystem": [],
"SearchParameter": [],
"OperationDefinition": [],
"CapabilityStatement": [],
"examples": []
}
resource_files = self.get_resource_files(package_id, version)
for file_path in resource_files:
try:
with open(file_path, 'r') as f:
resource = json.load(f)
resource_type = resource.get("resourceType", "unknown")
if resource_type in index:
index[resource_type].append({
"id": resource.get("id"),
"url": resource.get("url"),
"name": resource.get("name"),
"title": resource.get("title"),
"version": resource.get("version"),
"file_path": str(file_path)
})
else:
# Treat as example
index["examples"].append({
"resourceType": resource_type,
"id": resource.get("id"),
"file_path": str(file_path)
})
except (json.JSONDecodeError, IOError) as e:
print(f"Error processing {file_path}: {e}")
continue
return index
def search_resources(self, package_id: str, version: str = "latest",
query: str = None, resource_type: str = None) -> List[Dict]:
"""Search for resources in package by name, url, or type"""
index = self.build_resource_index(package_id, version)
results = []
for res_type, resources in index.items():
if resource_type and res_type != resource_type:
continue
for resource in resources:
if not query:
results.append({**resource, "resourceType": res_type})
elif (query.lower() in str(resource.get("name", "")).lower() or
query.lower() in str(resource.get("url", "")).lower() or
query.lower() in str(resource.get("title", "")).lower()):
results.append({**resource, "resourceType": res_type})
return results
def main():
parser = argparse.ArgumentParser(description="FHIR Package Manager")
parser.add_argument("--cache-dir", help="Custom cache directory")
subparsers = parser.add_subparsers(dest="command", help="Commands")
# Install command
install_parser = subparsers.add_parser("install", help="Install a package")
install_parser.add_argument("package_id", help="Package ID (e.g., hl7.fhir.r4.core)")
install_parser.add_argument("--version", default="latest", help="Package version")
# List command
list_parser = subparsers.add_parser("list", help="List installed packages")
# Search command
search_parser = subparsers.add_parser("search", help="Search resources in package")
search_parser.add_argument("package_id", help="Package ID")
search_parser.add_argument("--query", help="Search query")
search_parser.add_argument("--type", help="Resource type filter")
search_parser.add_argument("--version", default="latest", help="Package version")
# Index command
index_parser = subparsers.add_parser("index", help="Build resource index for package")
index_parser.add_argument("package_id", help="Package ID")
index_parser.add_argument("--version", default="latest", help="Package version")
args = parser.parse_args()
if not args.command:
parser.print_help()
return
manager = FHIRPackageManager(args.cache_dir)
if args.command == "install":
try:
manager.install_package(args.package_id, args.version)
except Exception as e:
print(f"Error installing package: {e}", file=sys.stderr)
sys.exit(1)
elif args.command == "list":
packages = manager.list_installed()
if packages:
print("Installed packages:")
for pkg in packages:
print(f" {pkg['id']}@{pkg['version']} -> {pkg['path']}")
else:
print("No packages installed")
elif args.command == "search":
results = manager.search_resources(args.package_id, args.version,
args.query, args.type)
if results:
print(f"Found {len(results)} resources:")
for result in results:
name = result.get("name") or result.get("id", "unknown")
print(f" {result['resourceType']}: {name}")
if result.get("url"):
print(f" URL: {result['url']}")
else:
print("No resources found")
elif args.command == "index":
index = manager.build_resource_index(args.package_id, args.version)
print(f"Resource index for {args.package_id}@{args.version}:")
for res_type, resources in index.items():
if resources:
print(f" {res_type}: {len(resources)} resources")
if __name__ == "__main__":
main()
scripts/fhir_validator.ts#!/usr/bin/env node
/**
* FHIR Resource Validator and Processor
* Validates FHIR resources against profiles and provides formatting
*/
import { readFileSync, writeFileSync } from 'fs';
import { resolve } from 'path';
import * as yargs from 'yargs';
interface FHIRResource {
resourceType: string;
id?: string;
meta?: {
profile?: string[];
versionId?: string;
lastUpdated?: string;
};
[key: string]: any;
}
interface ValidationResult {
valid: boolean;
errors: string[];
warnings: string[];
}
class FHIRValidator {
private knownResourceTypes = new Set([
'Patient', 'Practitioner', 'Organization', 'Location', 'Observation',
'DiagnosticReport', 'Medication', 'MedicationRequest', 'AllergyIntolerance',
'Condition', 'Procedure', 'Encounter', 'Appointment', 'Bundle',
'CapabilityStatement', 'StructureDefinition', 'ValueSet', 'CodeSystem'
]);
validateResource(resource: FHIRResource): ValidationResult {
const result: ValidationResult = {
valid: true,
errors: [],
warnings: []
};
// Basic structure validation
if (!resource.resourceType) {
result.errors.push('Missing required field: resourceType');
result.valid = false;
} else if (!this.knownResourceTypes.has(resource.resourceType)) {
result.warnings.push(`Unknown resource type: ${resource.resourceType}`);
}
// ID validation
if (resource.id && !/^[A-Za-z0-9\-\.]{1,64}$/.test(resource.id)) {
result.errors.push('Invalid ID format - must be 1-64 characters, alphanumeric, hyphens, dots only');
result.valid = false;
}
// Resource-specific validation
switch (resource.resourceType) {
case 'Patient':
this.validatePatient(resource, result);
break;
case 'Observation':
this.validateObservation(resource, result);
break;
case 'Bundle':
this.validateBundle(resource, result);
break;
}
return result;
}
private validatePatient(patient: FHIRResource, result: ValidationResult): void {
// Patient.name validation
if (patient.name && Array.isArray(patient.name)) {
patient.name.forEach((name: any, index: number) => {
if (!name.family && !name.given) {
result.warnings.push(`Patient.name[${index}]: Should have either family or given name`);
}
});
}
// Gender validation
if (patient.gender && !['male', 'female', 'other', 'unknown'].includes(patient.gender)) {
result.errors.push(`Invalid gender value: ${patient.gender}`);
result.valid = false;
}
// Birth date validation
if (patient.birthDate && !/^\d{4}-\d{2}-\d{2}$/.test(patient.birthDate)) {
result.errors.push('Invalid birthDate format - must be YYYY-MM-DD');
result.valid = false;
}
}
private validateObservation(observation: FHIRResource, result: ValidationResult): void {
// Status is required
if (!observation.status) {
result.errors.push('Observation.status is required');
result.valid = false;
} else if (!['registered', 'preliminary', 'final', 'amended', 'corrected', 'cancelled', 'entered-in-error', 'unknown'].includes(observation.status)) {
result.errors.push(`Invalid status: ${observation.status}`);
result.valid = false;
}
// Code is required
if (!observation.code) {
result.errors.push('Observation.code is required');
result.valid = false;
}
// Subject reference validation
if (!observation.subject) {
result.errors.push('Observation.subject is required');
result.valid = false;
} else if (observation.subject.reference && !observation.subject.reference.includes('/')) {
result.warnings.push('Subject reference should include resource type (e.g., Patient/123)');
}
}
private validateBundle(bundle: FHIRResource, result: ValidationResult): void {
// Type is required
if (!bundle.type) {
result.errors.push('Bundle.type is required');
result.valid = false;
}
// Entry validation
if (bundle.entry && Array.isArray(bundle.entry)) {
bundle.entry.forEach((entry: any, index: number) => {
if (entry.resource) {
const entryValidation = this.validateResource(entry.resource);
if (!entryValidation.valid) {
result.errors.push(`Bundle.entry[${index}]: ${entryValidation.errors.join(', ')}`);
result.valid = false;
}
}
});
}
}
formatResource(resource: FHIRResource, options: { indent?: number } = {}): string {
const indent = options.indent || 2;
// Ensure proper ordering of common FHIR fields
const ordered: any = {};
// Standard order for FHIR resources
const fieldOrder = ['resourceType', 'id', 'meta', 'identifier', 'active', 'name', 'status', 'code', 'subject', 'entry'];
fieldOrder.forEach(field => {
if (resource[field] !== undefined) {
ordered[field] = resource[field];
}
});
// Add remaining fields
Object.keys(resource).forEach(key => {
if (!fieldOrder.includes(key)) {
ordered[key] = resource[key];
}
});
return JSON.stringify(ordered, null, indent);
}
generateExample(resourceType: string): FHIRResource | null {
const examples: Record<string, FHIRResource> = {
Patient: {
resourceType: 'Patient',
id: 'example',
active: true,
name: [{
family: 'Doe',
given: ['John']
}],
gender: 'male',
birthDate: '1990-01-01'
},
Observation: {
resourceType: 'Observation',
id: 'example',
status: 'final',
code: {
coding: [{
system: 'http://loinc.org',
code: '55284-4',
display: 'Blood pressure systolic & diastolic'
}]
},
subject: {
reference: 'Patient/example'
},
valueQuantity: {
value: 120,
unit: 'mmHg',
system: 'http://unitsofmeasure.org',
code: 'mm[Hg]'
}
},
Bundle: {
resourceType: 'Bundle',
id: 'example',
type: 'collection',
entry: [{
resource: {
resourceType: 'Patient',
id: 'patient1',
active: true,
name: [{
family: 'Example',
given: ['Patient']
}]
}
}]
}
};
return examples[resourceType] || null;
}
}
// CLI Interface
async function main() {
const argv = yargs
.command('validate <file>', 'Validate a FHIR resource file', (yargs) => {
yargs.positional('file', {
describe: 'Path to FHIR resource JSON file',
type: 'string'
});
})
.command('format <file>', 'Format a FHIR resource file', (yargs) => {
yargs
.positional('file', {
describe: 'Path to FHIR resource JSON file',
type: 'string'
})
.option('output', {
alias: 'o',
describe: 'Output file path',
type: 'string'
})
.option('indent', {
alias: 'i',
describe: 'Indentation spaces',
type: 'number',
default: 2
});
})
.command('example <resourceType>', 'Generate example FHIR resource', (yargs) => {
yargs.positional('resourceType', {
describe: 'FHIR resource type',
type: 'string'
});
})
.demandCommand()
.help()
.argv;
const validator = new FHIRValidator();
try {
if (argv._[0] === 'validate') {
const filePath = resolve(argv.file as string);
const content = readFileSync(filePath, 'utf8');
const resource: FHIRResource = JSON.parse(content);
const result = validator.validateResource(resource);
console.log(`\nValidating ${resource.resourceType}/${resource.id || 'unknown'}:`);
if (result.valid) {
console.log('✅ Valid FHIR resource');
} else {
console.log('❌ Invalid FHIR resource');
result.errors.forEach(error => console.log(` Error: ${error}`));
}
if (result.warnings.length > 0) {
console.log('\nWarnings:');
result.warnings.forEach(warning => console.log(` Warning: ${warning}`));
}
process.exit(result.valid ? 0 : 1);
}
else if (argv._[0] === 'format') {
const filePath = resolve(argv.file as string);
const content = readFileSync(filePath, 'utf8');
const resource: FHIRResource = JSON.parse(content);
const formatted = validator.formatResource(resource, { indent: argv.indent as number });
if (argv.output) {
writeFileSync(resolve(argv.output as string), formatted);
console.log(`Formatted resource written to ${argv.output}`);
} else {
console.log(formatted);
}
}
else if (argv._[0] === 'example') {
const resourceType = argv.resourceType as string;
const example = validator.generateExample(resourceType);
if (example) {
const formatted = validator.formatResource(example);
console.log(formatted);
} else {
console.log(`No example available for resource type: ${resourceType}`);
process.exit(1);
}
}
} catch (error) {
console.error('Error:', error.message);
process.exit(1);
}
}
if (require.main === module) {
main();
}
export { FHIRValidator };