Async/sync FHIR client for Python providing comprehensive API for CRUD operations over FHIR resources
68
FHIR resource instances with full CRUD capabilities, validation, serialization, and path-based data access. Resources act as dictionary-like objects with additional FHIR-specific functionality.
Core operations for managing individual FHIR resources on the server.
async def create(self, **kwargs):
"""
Create the resource on the server.
Parameters:
- **kwargs: Additional parameters for creation (e.g., conditional create)
Returns:
Self with server-assigned ID and metadata
Raises:
- OperationOutcome: If creation fails
"""
async def update(self, **kwargs):
"""
Update the existing resource on the server.
Parameters:
- **kwargs: Additional parameters for update
Returns:
Self with updated data and metadata
Raises:
- ResourceNotFound: If resource doesn't exist
- OperationOutcome: If update fails
"""
async def patch(self, **kwargs):
"""
Patch the existing resource on the server.
Parameters:
- **kwargs: Additional parameters for patch operation
Returns:
Self with patched data and metadata
Raises:
- ResourceNotFound: If resource doesn't exist
- OperationOutcome: If patch fails
"""
async def delete(self):
"""
Delete the resource from the server.
Raises:
- ResourceNotFound: If resource doesn't exist
- OperationOutcome: If deletion fails
"""
async def save(self, **kwargs):
"""
Save the resource (create if new, update if existing).
Parameters:
- **kwargs: Additional parameters for save operation
Returns:
Self with saved data and metadata
"""
async def refresh(self):
"""
Refresh resource data from the server.
Returns:
Self with current server data
Raises:
- ResourceNotFound: If resource no longer exists
"""Validate resources against FHIR profiles and business rules.
async def is_valid(self) -> bool:
"""
Validate resource using server $validate operation.
Returns:
True if resource is valid, False otherwise
"""Convert resources to different formats and create references.
def to_reference(self, **kwargs) -> FHIRReference:
"""
Convert resource to a reference.
Parameters:
- **kwargs: Additional reference attributes (display, type, etc.)
Returns:
Reference pointing to this resource
"""
def serialize(self) -> dict:
"""
Serialize resource to a plain dictionary.
Returns:
Dict representation suitable for JSON serialization
"""Access nested resource data using dot-notation paths.
def get_by_path(self, path: str, default=None):
"""
Get value from resource using dot-separated path.
Parameters:
- path: Dot-separated path (e.g., 'name.0.family')
- default: Value to return if path not found
Returns:
Value at path or default
"""Core properties for accessing resource metadata.
@property
def resource_type(self) -> str:
"""The FHIR resource type (e.g., 'Patient', 'Observation')"""
@property
def id(self) -> str:
"""The resource identifier assigned by the server"""
@property
def reference(self) -> str:
"""Full reference string in format 'ResourceType/id'"""import asyncio
from fhirpy import AsyncFHIRClient
async def basic_operations():
client = AsyncFHIRClient('https://hapi.fhir.org/baseR4')
# Create a new patient
patient = client.resource('Patient',
name=[{
'family': 'Doe',
'given': ['John', 'William']
}],
gender='male',
birthDate='1990-01-01',
active=True
)
# Save to server (create)
await patient.create()
print(f"Created patient: {patient.id}")
# Update patient
patient['telecom'] = [{
'system': 'email',
'value': 'john.doe@example.com'
}]
await patient.save()
# Refresh from server
await patient.refresh()
# Delete patient
await patient.delete()async def conditional_operations():
client = AsyncFHIRClient('https://hapi.fhir.org/baseR4')
# Conditional create - only create if identifier doesn't exist
patient = client.resource('Patient',
identifier=[{
'system': 'http://example.org/ids',
'value': 'unique-id-123'
}],
name=[{'family': 'Smith', 'given': ['Jane']}]
)
await patient.create(ifNoneExist='identifier=unique-id-123')
# Conditional update - update only if version matches
patient['active'] = False
await patient.update(ifMatch='W/"1"')async def validation_example():
client = AsyncFHIRClient('https://hapi.fhir.org/baseR4')
# Create patient with invalid data
patient = client.resource('Patient',
name=[{'family': ''}], # Empty family name
gender='invalid-gender' # Invalid gender value
)
# Validate before saving
is_valid = await patient.is_valid()
if not is_valid:
print("Patient has validation errors")
# Fix issues before saving
patient['name'][0]['family'] = 'ValidName'
patient['gender'] = 'unknown'
if await patient.is_valid():
await patient.create()async def path_access_example():
client = AsyncFHIRClient('https://hapi.fhir.org/baseR4')
patient = client.resource('Patient',
name=[
{
'family': 'Doe',
'given': ['John', 'William'],
'use': 'official'
},
{
'family': 'Smith',
'given': ['Johnny'],
'use': 'nickname'
}
],
address=[{
'line': ['123 Main St', 'Apt 4B'],
'city': 'Springfield',
'postalCode': '12345',
'country': 'US'
}]
)
# Access nested data using paths
family_name = patient.get_by_path('name.0.family') # 'Doe'
first_given = patient.get_by_path('name.0.given.0') # 'John'
city = patient.get_by_path('address.0.city') # 'Springfield'
street = patient.get_by_path('address.0.line.0') # '123 Main St'
# Safe access with defaults
phone = patient.get_by_path('telecom.0.value', 'No phone') # 'No phone'
print(f"Patient: {first_given} {family_name} from {city}")async def reference_example():
client = AsyncFHIRClient('https://hapi.fhir.org/baseR4')
# Create patient
patient = client.resource('Patient',
name=[{'family': 'Doe', 'given': ['John']}]
)
await patient.create()
# Create observation referencing the patient
observation = client.resource('Observation',
status='final',
code={
'coding': [{
'system': 'http://loinc.org',
'code': '55284-4',
'display': 'Blood pressure'
}]
},
subject=patient.to_reference(), # Convert to reference
valueQuantity={
'value': 120,
'unit': 'mmHg',
'system': 'http://unitsofmeasure.org',
'code': 'mm[Hg]'
}
)
await observation.create()
# Access reference properties
patient_ref = observation['subject']
print(f"Subject reference: {patient_ref.reference}")
print(f"Subject ID: {patient_ref.id}")
print(f"Subject type: {patient_ref.resource_type}")async def serialization_example():
client = AsyncFHIRClient('https://hapi.fhir.org/baseR4')
patient = client.resource('Patient',
name=[{'family': 'Doe', 'given': ['John']}],
active=True
)
# Serialize for storage or transmission
patient_dict = patient.serialize()
# Convert to JSON string
import json
patient_json = json.dumps(patient_dict, indent=2)
print(patient_json)
# Load from dict
loaded_patient = client.resource('Patient', **patient_dict)Install with Tessl CLI
npx tessl i tessl/pypi-fhirpyevals
scenario-1
scenario-2
scenario-3
scenario-4
scenario-5
scenario-6
scenario-7
scenario-8
scenario-9
scenario-10