Async/sync FHIR client for Python providing comprehensive API for CRUD operations over FHIR resources
68
Helper functions and classes for data manipulation, parameter transformation, and path-based access. These utilities support common operations needed when working with FHIR data structures.
Enhanced dictionary and list classes with path-based access capabilities.
class AttrDict(dict):
"""Dictionary with attribute-style access and path-based operations."""
def __init__(self, *args, **kwargs):
"""Initialize AttrDict from dict or keyword arguments."""
def get_by_path(self, path: str, default=None):
"""
Get value using dot-separated path.
Parameters:
- path: Dot-separated path string (e.g., 'address.0.city')
- default: Value to return if path not found
Returns:
Value at path or default
"""
class SearchList(list):
"""List with path-based access capabilities."""
def get_by_path(self, path: str, default=None):
"""
Get value using dot-separated path.
Parameters:
- path: Dot-separated path string
- default: Value to return if path not found
Returns:
Value at path or default
"""Functions for working with collections and data processing.
def chunks(lst: list, n: int) -> Generator:
"""
Split list into chunks of specified size.
Parameters:
- lst: List to split
- n: Chunk size
Yields:
Sublists of size n (last chunk may be smaller)
"""
def unique_everseen(seq: list) -> list:
"""
Get unique items from sequence, preserving order.
Parameters:
- seq: Input sequence
Returns:
List with unique items in original order
"""Functions for parsing and manipulating dot-separated paths.
def parse_path(path: str) -> list:
"""
Parse dot-separated path into list of keys.
Parameters:
- path: Dot-separated path string
Returns:
List of path components
"""
def get_by_path(obj, path: list, default=None):
"""
Get value from nested object using path list.
Parameters:
- obj: Object to traverse
- path: List of keys/indices
- default: Value to return if path not found
Returns:
Value at path or default
"""
def set_by_path(obj, path: str, value):
"""
Set value in nested object using dot-separated path.
Parameters:
- obj: Object to modify
- path: Dot-separated path string
- value: Value to set
"""
def convert_values(data, fn):
"""
Recursively convert data values using provided function.
Parameters:
- data: Data structure to convert (dict, list, or other)
- fn: Function that takes a value and returns (converted_value, stop_flag)
Returns:
Converted data structure with fn applied recursively
"""Functions for encoding parameters and working with URLs.
def encode_params(params: dict) -> str:
"""
Encode parameters for URL query string.
Parameters:
- params: Dictionary of parameters
Returns:
URL-encoded query string
"""Functions for formatting dates and values according to FHIR specifications.
def format_date_time(date: datetime) -> str:
"""
Format datetime for FHIR (ISO 8601 format).
Parameters:
- date: Python datetime object
Returns:
FHIR-formatted datetime string (YYYY-MM-DDTHH:MM:SSZ)
"""
def format_date(date: date) -> str:
"""
Format date for FHIR.
Parameters:
- date: Python date object
Returns:
FHIR-formatted date string (YYYY-MM-DD)
"""
def transform_param(param: str) -> str:
"""
Transform parameter name for FHIR (underscore to dash).
Parameters:
- param: Parameter name with underscores
Returns:
Parameter name with dashes (preserves leading underscore/dot)
"""
def transform_value(value) -> str:
"""
Transform value for FHIR search parameters.
Handles:
- datetime objects → ISO format strings
- date objects → YYYY-MM-DD strings
- bool → 'true'/'false' strings
- BaseResource/BaseReference → reference strings
Parameters:
- value: Value to transform
Returns:
String representation suitable for FHIR
"""from fhirpy.base.utils import AttrDict, get_by_path, set_by_path, parse_path
def path_access_examples():
# AttrDict with attribute access
patient_data = AttrDict({
'name': [
{'family': 'Doe', 'given': ['John', 'William']},
{'family': 'Smith', 'given': ['Johnny'], 'use': 'nickname'}
],
'address': [{
'line': ['123 Main St', 'Apt 4B'],
'city': 'Springfield',
'postalCode': '12345'
}],
'telecom': [
{'system': 'phone', 'value': '555-1234'},
{'system': 'email', 'value': 'john@example.com'}
]
})
# Access using paths
family_name = patient_data.get_by_path('name.0.family') # 'Doe'
first_given = patient_data.get_by_path('name.0.given.0') # 'John'
street = patient_data.get_by_path('address.0.line.0') # '123 Main St'
phone = patient_data.get_by_path('telecom.0.value') # '555-1234'
# Safe access with defaults
fax = patient_data.get_by_path('telecom.2.value', 'No fax') # 'No fax'
print(f"Patient: {first_given} {family_name}")
print(f"Address: {street}, {patient_data.get_by_path('address.0.city')}")
# Attribute-style access
print(f"Name array: {patient_data.name}")
print(f"City: {patient_data.address[0]['city']}")from fhirpy.base.utils import chunks, unique_everseen, set_by_path
def data_manipulation_examples():
# Split large datasets into manageable chunks
patient_ids = [f"patient-{i}" for i in range(1, 101)] # 100 patient IDs
# Process in chunks of 20
for chunk in chunks(patient_ids, 20):
print(f"Processing batch of {len(chunk)} patients")
# Process chunk...
# Remove duplicates while preserving order
search_params = ['name', 'birthdate', 'gender', 'name', 'active', 'gender']
unique_params = unique_everseen(search_params)
print(f"Unique parameters: {unique_params}") # ['name', 'birthdate', 'gender', 'active']
# Modify nested data structures
patient_data = {'name': [{'family': 'Doe'}]}
set_by_path(patient_data, 'name.0.given', ['John'])
set_by_path(patient_data, 'active', True)
set_by_path(patient_data, 'address.0.city', 'Springfield') # Creates nested structure
print(f"Modified patient: {patient_data}")import datetime
from fhirpy.base.utils import transform_param, transform_value, format_date_time, format_date
def parameter_transformation_examples():
# Parameter name transformation
fhir_param = transform_param('general_practitioner') # 'general-practitioner'
preserved = transform_param('_id') # '_id' (unchanged)
dot_param = transform_param('.effectiveDate') # '.effectiveDate' (unchanged)
print(f"Transformed: {fhir_param}")
# Value transformation for search parameters
now = datetime.datetime(2024, 1, 15, 10, 30, 0, tzinfo=datetime.timezone.utc)
today = datetime.date(2024, 1, 15)
datetime_str = transform_value(now) # '2024-01-15T10:30:00Z'
date_str = transform_value(today) # '2024-01-15'
bool_str = transform_value(True) # 'true'
false_str = transform_value(False) # 'false'
regular_str = transform_value('text') # 'text'
print(f"DateTime: {datetime_str}")
print(f"Date: {date_str}")
print(f"Boolean: {bool_str}")
# Direct date formatting
fhir_datetime = format_date_time(now) # '2024-01-15T10:30:00Z'
fhir_date = format_date(today) # '2024-01-15'from fhirpy.base.utils import encode_params
def parameter_encoding_examples():
# Simple parameters
params1 = {'name': ['Smith'], 'active': ['true']}
query1 = encode_params(params1) # 'name=Smith&active=true'
# Multiple values for same parameter
params2 = {'status:not': ['active', 'entered-in-error']}
query2 = encode_params(params2) # 'status:not=active&status:not=entered-in-error'
# Complex parameters with modifiers
params3 = {
'name:contains': ['John'],
'birthdate:ge': ['1990-01-01'],
'address-city': ['Springfield', 'Boston']
}
query3 = encode_params(params3)
print(f"Simple: {query1}")
print(f"Multiple: {query2}")
print(f"Complex: {query3}")from fhirpy.base.utils import AttrDict, SearchList
import json
async def search_results_processing():
# Simulate FHIR Bundle response
bundle_data = {
'resourceType': 'Bundle',
'total': 3,
'entry': [
{
'resource': {
'resourceType': 'Patient',
'id': '1',
'name': [{'family': 'Doe', 'given': ['John']}],
'address': [{'city': 'Springfield'}]
}
},
{
'resource': {
'resourceType': 'Patient',
'id': '2',
'name': [{'family': 'Smith', 'given': ['Jane']}],
'address': [{'city': 'Boston'}]
}
}
]
}
# Convert to AttrDict for easier access
bundle = AttrDict(bundle_data)
# Extract patient data with path operations
patients = SearchList()
for entry in bundle.entry:
patient = AttrDict(entry.resource)
patients.append(patient)
# Process patients using path-based access
for patient in patients:
name = patient.get_by_path('name.0.family')
given = patient.get_by_path('name.0.given.0')
city = patient.get_by_path('address.0.city', 'Unknown')
print(f"Patient {patient.id}: {given} {name} from {city}")
# Aggregate data
cities = [p.get_by_path('address.0.city') for p in patients if p.get_by_path('address.0.city')]
unique_cities = unique_everseen(cities)
print(f"Cities represented: {unique_cities}")from fhirpy import AsyncFHIRClient
from fhirpy.base.utils import chunks, transform_value
import datetime
async def utility_integration_example():
client = AsyncFHIRClient('https://hapi.fhir.org/baseR4')
# Batch process large number of patients
patient_ids = [f"patient-{i}" for i in range(1, 101)]
all_patients = []
for id_chunk in chunks(patient_ids, 10): # Process 10 at a time
# Build search for this chunk
id_query = '|'.join(id_chunk) # FHIR OR syntax
patients = await (client.resources('Patient')
.search(_id=id_query)
.fetch())
all_patients.extend(patients)
print(f"Processed {len(all_patients)} patients")
# Use date transformation for search
cutoff_date = datetime.date(2020, 1, 1)
recent_patients = await (client.resources('Patient')
.search(birthdate__ge=transform_value(cutoff_date))
.fetch())
print(f"Found {len(recent_patients)} patients born after {cutoff_date}")
# Process results with path operations
for patient in recent_patients:
birth_date = patient.get_by_path('birthDate')
name = patient.get_by_path('name.0.family', 'Unknown')
print(f"{name}: {birth_date}")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