Async/sync FHIR client for Python providing comprehensive API for CRUD operations over FHIR resources
68
FHIR reference handling with resolution capabilities and local/external reference detection. References represent relationships between FHIR resources and can be resolved to retrieve the referenced resource.
Resolve references to retrieve the actual referenced resources.
async def resolve(self) -> FHIRResource:
"""
Resolve the reference to retrieve the referenced resource.
Returns:
The resource instance that this reference points to
Raises:
- ResourceNotFound: If referenced resource doesn't exist
- InvalidResponse: If reference cannot be resolved
"""Properties for accessing reference information and metadata.
@property
def reference(self) -> str:
"""
The reference string (e.g., 'Patient/123' or 'https://external.com/Patient/456').
Returns:
Full reference string
"""
@property
def id(self) -> str:
"""
The resource ID if this is a local reference.
Returns:
Resource ID for local references, None for external references
"""
@property
def resource_type(self) -> str:
"""
The resource type if this is a local reference.
Returns:
Resource type for local references, None for external references
"""
@property
def is_local(self) -> bool:
"""
Whether this reference points to a local resource.
Returns:
True for local references (ResourceType/id), False for external URLs
"""import asyncio
from fhirpy import AsyncFHIRClient
async def creating_references():
client = AsyncFHIRClient('https://hapi.fhir.org/baseR4')
# Create reference by resource type and ID
patient_ref = client.reference('Patient', '123')
print(f"Reference: {patient_ref.reference}") # 'Patient/123'
# Create reference by full reference string
external_ref = client.reference(reference='https://external.com/Patient/456')
print(f"External reference: {external_ref.reference}")
# Create reference with additional attributes
display_ref = client.reference('Patient', '123',
display='John Doe',
type='Patient')
# Convert resource to reference
patient = client.resource('Patient', id='123',
name=[{'family': 'Doe', 'given': ['John']}])
patient_ref = patient.to_reference(display='John Doe')async def reference_properties():
client = AsyncFHIRClient('https://hapi.fhir.org/baseR4')
# Local reference
local_ref = client.reference('Patient', '123')
print(f"Is local: {local_ref.is_local}") # True
print(f"Resource type: {local_ref.resource_type}") # 'Patient'
print(f"ID: {local_ref.id}") # '123'
print(f"Reference: {local_ref.reference}") # 'Patient/123'
# External reference
external_ref = client.reference(reference='https://external.com/Patient/456')
print(f"Is local: {external_ref.is_local}") # False
print(f"Resource type: {external_ref.resource_type}") # None
print(f"ID: {external_ref.id}") # None
print(f"Reference: {external_ref.reference}") # 'https://external.com/Patient/456'
# Reference with only display (contained resource scenario)
display_only_ref = client.reference(reference='#patient1',
display='Contained Patient')
print(f"Is local: {display_only_ref.is_local}") # False (not ResourceType/id format)async def reference_resolution():
client = AsyncFHIRClient('https://hapi.fhir.org/baseR4')
# Create a patient first
patient = client.resource('Patient',
name=[{'family': 'Doe', 'given': ['John']}])
await patient.create()
# Create reference to the patient
patient_ref = client.reference('Patient', patient.id)
# Resolve reference to get full patient data
resolved_patient = await patient_ref.resolve()
print(f"Resolved patient name: {resolved_patient.get_by_path('name.0.family')}")
# Use reference in another resource
observation = client.resource('Observation',
status='final',
code={
'coding': [{
'system': 'http://loinc.org',
'code': '55284-4',
'display': 'Blood pressure'
}]
},
subject=patient_ref, # Reference to patient
valueQuantity={
'value': 120,
'unit': 'mmHg'
}
)
await observation.create()
# Later, resolve the subject reference
subject_ref = observation['subject']
subject_patient = await subject_ref.resolve()
print(f"Subject: {subject_patient.get_by_path('name.0.given.0')}")async def references_in_resources():
client = AsyncFHIRClient('https://hapi.fhir.org/baseR4')
# Create related resources
practitioner = client.resource('Practitioner',
name=[{'family': 'Smith', 'given': ['Dr. Jane']}])
await practitioner.create()
organization = client.resource('Organization',
name='General Hospital')
await organization.create()
patient = client.resource('Patient',
name=[{'family': 'Doe', 'given': ['John']}],
generalPractitioner=[practitioner.to_reference()],
managingOrganization=organization.to_reference()
)
await patient.create()
# Access and resolve references
gp_ref = patient['generalPractitioner'][0] # First GP reference
org_ref = patient['managingOrganization'] # Organization reference
print(f"GP reference: {gp_ref.reference}")
print(f"Org reference: {org_ref.reference}")
# Resolve references to get full data
gp = await gp_ref.resolve()
org = await org_ref.resolve()
print(f"GP Name: {gp.get_by_path('name.0.family')}")
print(f"Organization: {org['name']}")async def reference_error_handling():
client = AsyncFHIRClient('https://hapi.fhir.org/baseR4')
# Create reference to non-existent resource
invalid_ref = client.reference('Patient', 'non-existent-id')
try:
resolved = await invalid_ref.resolve()
except ResourceNotFound:
print("Referenced resource does not exist")
# Check reference validity before resolving
def is_valid_reference_format(ref):
if ref.is_local:
return ref.resource_type and ref.id
return ref.reference.startswith(('http://', 'https://'))
refs_to_check = [
client.reference('Patient', '123'),
client.reference(reference='https://external.com/Patient/456'),
client.reference(reference='#contained1')
]
for ref in refs_to_check:
if is_valid_reference_format(ref):
try:
resolved = await ref.resolve()
print(f"Successfully resolved: {ref.reference}")
except ResourceNotFound:
print(f"Valid format but resource not found: {ref.reference}")
else:
print(f"Invalid reference format: {ref.reference}")async def bundle_and_contained_references():
client = AsyncFHIRClient('https://hapi.fhir.org/baseR4')
# Create bundle with references between entries
bundle_data = {
'resourceType': 'Bundle',
'type': 'transaction',
'entry': [
{
'fullUrl': 'urn:uuid:patient-1',
'resource': {
'resourceType': 'Patient',
'name': [{'family': 'Doe', 'given': ['John']}]
},
'request': {'method': 'POST', 'url': 'Patient'}
},
{
'fullUrl': 'urn:uuid:observation-1',
'resource': {
'resourceType': 'Observation',
'status': 'final',
'code': {
'coding': [{
'system': 'http://loinc.org',
'code': '55284-4'
}]
},
'subject': {'reference': 'urn:uuid:patient-1'}, # Bundle reference
'valueQuantity': {'value': 120, 'unit': 'mmHg'}
},
'request': {'method': 'POST', 'url': 'Observation'}
}
]
}
# Submit bundle
result = await client.execute('', method='post', data=bundle_data)
# Work with contained resources
patient_with_contained = client.resource('Patient',
name=[{'family': 'Parent', 'given': ['Jane']}],
contained=[{
'resourceType': 'RelatedPerson',
'id': 'child1',
'patient': {'reference': '#'}, # Reference to containing resource
'relationship': [{
'coding': [{
'system': 'http://terminology.hl7.org/CodeSystem/v3-RoleCode',
'code': 'CHILD'
}]
}]
}],
contact=[{
'relationship': [{
'coding': [{
'system': 'http://terminology.hl7.org/CodeSystem/v2-0131',
'code': 'C'
}]
}],
'name': {'family': 'Child', 'given': ['Little']},
# Reference to contained resource
'organization': {'reference': '#child1'}
}]
)
await patient_with_contained.create()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