async/sync FHIR client for python3. This package provides an API for CRUD operations over FHIR resources
pip install fhirpy
or to install the latest dev version:
pip install git+https://github.com/beda-software/fhir-py.git
You can test this library by interactive FHIR course in the repository Aidbox/jupyter-course.
import asyncio
from fhirpy import AsyncFHIRClient
async def main():
# Create an instance
client = AsyncFHIRClient(
'http://fhir-server/',
authorization='Bearer TOKEN',
)
# Search for patients
resources = client.resources('Patient') # Return lazy search set
resources = resources.search(name='John').limit(10).sort('name')
patients = await resources.fetch() # Returns list of AsyncFHIRResource
# Create Organization resource
organization = client.resource(
'Organization',
name='beda.software',
active=False
)
await organization.save()
# Update (PATCH) organization. Resource support accessing its elements
# both as attribute and as a dictionary keys
if organization['active'] is False:
organization.active = True
await organization.save(fields=['active'])
# `await organization.patch(active=True)` would do the same PATCH operation
# Get patient resource by reference and delete
patient_ref = client.reference('Patient', 'new_patient')
# Get resource from this reference
# (throw ResourceNotFound if no resource was found)
patient_res = await patient_ref.to_resource()
await patient_res.delete()
# Iterate over search set
org_resources = client.resources('Organization')
# Lazy loading resources page by page with page count = 100
async for org_resource in org_resources.limit(100):
print(org_resource.serialize())
if __name__ == '__main__':
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
patients = client.resources('Patient')
patients.search(birthdate__gt='1944', birthdate__lt='1964')
# /Patient?birthdate=gt1944&birthdate=lt1964
patients.search(name__contains='John')
# /Patient?name:contains=John
patients.search(name=['John', 'Rivera'])
# /Patient?name=John&name=Rivera
patients.search(name='John,Eva')
# /Patient?name=John,Eva
patients.search(family__exact='Moore')
# /Patient?family:exact=Moore
patients.search(address_state='TX')
# /Patient?address-state=TX
patients.search(active=True, _id='id')
# /Patient?active=true&_id=id
patients.search(gender__not=['male', 'female'])
# /Patient?gender:not=male&gender:not=female
patients.search(general_practitioner__Organization__name='Hospital')
# /Patient?general-practitioner:Organization.name=Hospital
patients.search(general_practitioner__name='Hospital')
# /Patient?general-practitioner.name=Hospital
practitioner = client.resources('Practitioner').search(_id='john-smith').first()
patients.search(general_practitioner=practitioner)
# /Patient?general-practitioner=Practitioner/john-smith
import pytz
import datetime
patients.search(birthdate__lt=datetime.datetime.now(pytz.utc))
# /Patient?birthdate=lt2019-11-19T20:16:08Z
patients.search(birthdate__gt=datetime.datetime(2013, 10, 27, tzinfo=pytz.utc))
# /Patient?birthdate=gt2013-10-27T00:00:00Z
conditions = client.resources('Condition')
conditions.search(code__text='headache')
# /Condition?code:text=headache
conditions.search(code__in='http://acme.org/fhir/ValueSet/cardiac-conditions')
# /Condition?code:in=http://acme.org/fhir/ValueSet/cardiac-conditions
conditions.search(code__not_in='http://acme.org/fhir/ValueSet/cardiac-conditions')
# /Condition?code:not-in=http://acme.org/fhir/ValueSet/cardiac-conditions
conditions.search(code__below='126851005')
# /Condition?code:below=126851005
conditions.search(code__above='126851005')
# /Condition?code:above=126851005
Sometimes you can find that fhir-py does not implement some search parameters from the FHIR specification.
In this case, you can use Raw() wrapper without any transformations
from fhirpy.base.searchset import Raw
patients = client.resources('Patient')
patients.search(Raw(**{'general-practitioner.name': 'Hospital'}))
# /Patient?general-practitioner.name=Hospital
Use reference to get resource by id
patient = await client.reference('Patient', '1').to_resource()
# /Patient/1
Or use FHIR search API with .first() or .get() as described below.
practitioners = client.resources('Practitioner')
try:
await practitioners.search(active=True, _id='id').get()
# /Practitioner?active=true&_id=id
except ResourceNotFound:
pass
except MultipleResourcesFound:
pass
await practitioners.search(name='Jack').first()
# /Practitioner?name=Jack&_count=1
await patients.sort('active', '-birthdate').first()
# /Patient?_sort=active,-birthdate&_count=1
await practitioners.search(active=True).count()
await patients.count()
await practitioners.fetch()
# /Practitioner
await patients.elements('name', 'telecom').fetch()
# /Patient?_elements=resourceType,name,id,telecom
Keep in mind that this method as well as .fetch() doesn't return any included resources. Use fetch_raw() if you want to get all included resources.
# Returns a list of `Practitioner` resources
await practitioners.search(address_city='Krasnoyarsk').fetch_all()
await patients.fetch_all()
# Get 100 resources
await practitioners.limit(100).fetch()
observations = client.resources('Observation')
observations.sort('status', '-date', 'category')
# /Observation?_sort=status,-date,category
# Get only specified set of elements for each resource
patients.elements('identifier', 'active', 'link')
# /Patient?_elements=identifier,active,link
# Get all elements except specified set
practitioners.elements('address', 'telecom', exclude=True)
result = await client.resources('EpisodeOfCare') \
.include('EpisodeOfCare', 'patient').fetch_raw()
# /EpisodeOfCare?_include=EpisodeOfCare:patient
for entry in result.entry:
print(entry.resource)
await client.resources('MedicationRequest') \
.include('MedicationRequest', 'patient', target_resource_type='Patient') \
.fetch_raw()
# /MedicationRequest?_include=MedicationRequest:patient:Patient
# For FHIR version >= 3.5 we can also use modifier :iterate
await client.resources('MedicationRequest') \
.include('MedicationDispense', 'prescription') \
.include('MedicationRequest', 'performer', iterate=True) \
.fetch_raw()
# /MedicationRequest?_include=MedicationDispense:prescription
# &_include:iterate=MedicationRequest:performer
# For FHIR version 3.0-3.3 use modifier :recurse
await client.resources('MedicationDispense') \
.include('MedicationRequest', 'prescriber', recursive=True) \
.fetch_raw()
# /MedicationDispense?_include:recurse=MedicationRequest:prescriber
await client.resources('Encounter').include('*') \
.fetch_raw()
# /Encounter?_include=*
await practitioners.revinclude('Group', 'member').fetch_raw()
# /Practitioner?_revinclude=Group:member
or
await practitioners.include('Group', 'member', reverse=True).fetch_raw()
# /Practitioner?_revinclude=Group:member
await client.resources('EpisodeOfCare').revinclude('*') \
.fetch_raw()
# /EpisodeOfCare?_revinclude=*
# resource.create(search_params)
# executes POST /Patient?identifier=fhirpy
patient = client.resource("Patient",
identifier=[{"system": "http://example.com/env", "value": "fhirpy"}],
name=[{"text": "Mr. Smith"}],
)
await patient.create(identifier="other")
# searchset.get_or_create(resource)
# executes POST /Patient?identifier=fhirpy
patient, created = await client.resources("Patient").search(identifier="fhirpy").get_or_create(patient_to_save)
# no match -> created is True
# one match -> created is False, return existing resource
# multiple matches -> 412 'MultipleResourcesFound'
# resource, created: bool = searchset.patch(resource)
# executes PUT /Patient?identifier=fhirpy
patient_to_update = client.resource("Patient",
identifier=[{"system": "http://example.com/env", "value": "fhirpy"}],
active=False)
new_patient, created = await client.resources("Patient").search(identifier="fhirpy").update(patient_to_update)
# no match -> created is True
# one match -> created is False, the matched resource is overwritten
# multiple matches -> 412 'MultipleResourcesFound'
# patched_resource = searchset.patch(resource)
# executes PATCH /Patient?identifier=fhirpy
patient_to_patch = client.resource("Patient",
identifier=[{"system": "http://example.com/env", "value": "fhirpy"}],
name=[{"text": "Mr. Smith"}])
patched_patient = await client.resources("Patient").search(identifier="fhirpy").patch(patient_to_patch)
# no match -> 404 'ResourceNotFound'
# multiple matches -> 412 'MultipleResourcesFound'
response_data, status_code = await self.client.resources("Patient").search(identifier="abc").delete()
# no match -> status_code = 204 'No Content'
# one match -> status_code = 200 'OK'
# multiple matches -> status_code = 412 'MultipleResourcesFound' (implementation specific)
Third party typing data models might be used along with fhir-py.
The typing data models should match ResourceProtocol, e.g. have resourceType attribute, optional id and be iterable for serialization.
There's a third party repository fhir-py-types that is written on top of pydantic models is fully compatible with fhir-py.
fhir-py uses typehints in the codebase and it statically checked by mypy. Some interfaces that are described below designed in the way to properly infer the return type based on the model class or instance.
To instantiate a resource, simply use type model constructor, e.g.
patient = Patient(name=[HumanName(text='Patient')])
Client class provides CRUD methods that designed to work with typed models.
await client.create(patient) # returns Patient
await client.create(patient) # returns Patient
Smart helper that creates or updated the resource based on having id
await client.save(patient) # returns Patient
Also it supports overriding specific fields using patch:
await client.save(patient, fields=['identifier']) # returns Patient
Patch accepts different syntaxes for patching, there're two syntaxes for general usage, without inferred types:
await client.patch('Patient', 'id', name=[HumanName(text='Patient')]) # returns Any
await client.patch('Patient/id', name=[HumanName(text='Patient')]) # returns Any
And two types that infers type:
await client.patch(Patient, 'id', name=[HumanName(text='Patient')]) # returns Patient
await client.patch(patient, name=[HumanName(text='Patient')]) # returns Patient
Delete accepts different syntaxes for resource deletion, there're also syntaxes similar to patch, but without output type because delete usually returns nothing.
await client.delete('Patient', 'id')
await client.delete('Patient/id')
await client.delete(Patient, 'id')
await client.delete(patient)
For fetching single resource by resourceType and id:
ss = await client.get(Patient, 'id') # returns Patient
For fetching multiple resources, SearchSet needs to be instantiated using the model class as the first argument
ss = client.resources(Patient) # returns AsyncFHIRSearchSet[Patient]
await ss.fetch_all() # returns list[Patient]
In that case search set infers model type for all methods that described above in the sections about search sets, including data fetching and conditional CRUD.
try:
await client.resource('Patient', birthDate='date', custom_prop='123', telecom=True) \
.is_valid(raise_exception=True)
except OperationOutcome as e:
print('Error: {}'.format(e))
patient = client.resource('Patient', birthDate='1998-01-01')
if (await patient.is_valid()):
pass
patient = await client.resources('Patient').first()
# Work with the resource as a dictionary
patient_family = patient['name'][0]['family']
# Or access value by an attribute
patient_given_name = patient.name[0].given[0]
patient_postal = patient.get_by_path(['resource', 'address', 0, 'postalCode'])
# get_by_path can be also used on any nested attribute
patient_name = patient.name[0]
patient_fullname = '{} {}'.format(
patient_name.get_by_path(['given', 0]),
patient_name.get_by_path(['family'])
)
# Get identifier value by specified system or empty string
uid = patient.get_by_path([
'resource', 'identifier',
{'system':'http://example.com/identifier/uid'},
'value'
], '')
# Get base value amount or 0
invoice = await client.resources('Invoice').first()
base_value = invoice.get_by_path([
'totalPriceComponent',
{'type': 'base'},
'amount', 'value'], 0)
resource = {
"name": [{"given": ["Firstname"], "family": "Lastname"}],
}
set_by_path(resource, ["name", 0, "given", 0], "FirstnameUpdated")
# resource
# {"name": [{"given": ["FirstnameUpdated"], "family": "Lastname"}]}
# Returns resources as dict
patient = await client.reference('Patient', '1').to_resource()
patient.serialize()
# Or
await client.reference('Patient', '1').to_resource().serialize()
# {'resourceType': 'Patient', 'id': '1', 'meta': {'versionId': '1', 'lastUpdated': '2021-11-13T11:50:24.685719Z'}, ...}
Both async and sync clients have identical sets of classes and methods.
| Sync | Async | |
|---|---|---|
| Client | SyncFHIRClient | AsyncFHIRClient |
| SearchSet | SyncFHIRSearchSet | AsyncFHIRSearchSet |
| Resource | SyncFHIRResource | AsyncFHIRResource |
| Reference | SyncFHIRReference | AsyncFHIRReference |
Import library:
from fhirpy import AsyncFHIRClient
To create AsyncFHIRClient instance use:
AsyncFHIRClient(url, authorization='', extra_headers={})
Returns an instance of the connection to the server which provides:
AsyncFHIRReference to the resourceAsyncFHIRResource which described belowAsyncFHIRSearchSetAsyncFHIRSearchSet[T]async .execute(path, method='post', data=None, params=None) - returns a result of FHIR operationdata model methods:
async .get(resource_type: type[T], id) - returns T instance by resourceType/idasync .get(resource_type: type[T], reference) - returns T instance by referenceasync .get(resource_type: str, id) - gets instance by resourceType/idasync .get(reference: str) - gets instance by referenceasync .save(resource: T, fields=[]) - creates or updates or patches (with fields=[...]) T instanceasync .create(resource: T) - creates T instanceasync .update(resource: T) - updates T instanceasync .patch(resource: T, **kwargs) - patches T instanceasync .patch(resource_type: type[T], id, **kwargs) - patches instance by resourceType/id and returns T instanceasync .patch(resource_type: type[T], reference, **kwargs) - patches instance by reference and returns T instanceasync .patch(resource_type: str, id, **kwargs) - patches instance by resourceType/idasync .patch(reference: str, **kwargs) - patches instance by referenceasync .delete(resource: T) - deletes T instanceasync .delete(resource_type: type[T], id) - deletes resource by resourceType/idasync .delete(resource_type: type[T], reference) - deletes resource by referenceasync .delete(resource_type: str, id) - deletes instance by resourceType/idasync .delete(reference: str) - deletes instance by referenceSometimes you need more control over the way http request is made and provide additional aiohttp session's request parameters like ssl, proxy, cookies, timeout etc. It's possible by providing aiohttp_config dict for AsyncFHIRClient:
client = AsyncFHIRClient(
FHIR_SERVER_URL,
aiohttp_config={
"ssl": ssl.create_default_context(),
"timeout": aiohttp.ClientTimeout(total=100),
}
)
Be careful and don't override other request values like params, json, data, auth, because it'll interfere with the way fhir-py works and lead to an incorrect behavior.
provides:
async .save(fields=[]) - creates or updates or patches (with fields=[...]) resource instanceasync .create() - creates resource instanceasync .update() - updates resource instanceasync .patch(**kwargs) - patches resource instanceasync .delete() - deletes resource instanceasync .refresh() - reloads resource from a serverasync .to_reference(**kwargs) - returns AsyncFHIRReference for this resourceasync .execute(operation, method='post', data=None, params=None) - returns a result of FHIR operation on the resourceprovides:
async .to_resource() - returns AsyncFHIRResource for this referenceasync .execute(operation, method='post', data=None, params=None) - returns a result of FHIR operation on the resourceasync .patch(**kwargs) - patches resource instanceasync .delete() - deletes resource instanceprovides:
async .fetch() - makes query to the server and returns a list of Resource filtered by resource typeasync .fetch_all() - makes query to the server and returns a full list of Resource filtered by resource typeasync .fetch_raw() - makes query to the server and returns a raw Bundle Resourceasync .first() - returns Resource or Noneasync .get() - returns Resource or raises ResourceNotFound when no resource found or MultipleResourcesFound when more than one resource found (parameter 'id' is deprecated)async .count() - makes query to the server and returns the total number of resources that match the SearchSetasync .get_or_create(resource) - conditional createasync .update(resource) - conditional updateasync .patch(**kwargs) - conditional patchImport library:
from fhirpy import SyncFHIRClient
To create SyncFHIRClient instance use:
SyncFHIRClient(url, authorization='', extra_headers={})
Returns an instance of the connection to the server which provides:
SyncFHIRReference to the resourceSyncFHIRResource which described belowSyncFHIRSearchSetPass requests_config parameter to SyncFHIRClient if you want to provide additional parameters for a request like verify, cert, timeout etc.
client = SyncFHIRClient(
FHIR_SERVER_URL,
requests_config={
"verify": False,
"allow_redirects": True,
"timeout": 60,
}
)
Be careful and don't override other request values like params, json, data, headers, which may interfere with the way fhir-py works and lead to an incorrect behavior.
The same as AsyncFHIRResource but with sync methods
provides: The same as AsyncFHIRReference but with sync methods
The same as AsyncFHIRSearchSet but with sync methods
(need some test FHIR server to run with, e.g. https://docs.aidbox.app/installation/setup-aidbox.dev)
Clone this repository:
https://github.com/beda-software/fhir-py.git
Go to fhir-py folder and install dev dependencies:
cd fhir-py
pip install -r requirements.txt
If you've already installed fhir-py library and want to test the last changes, reinstall it by running python setup.py install (or uninstall pip uninstall fhirpy)
Provide ENV variables FHIR_SERVER_URL and FHIR_SERVER_AUTHORIZATION, or edit tests/config.py
Run pytest
If you've found any bugs or think that some part of fhir-py is not compatible with FHIR spec, feel free to create an issue/pull request.
FAQs
FHIR client for python
We found that fhirpy demonstrated a healthy version release cadence and project activity because the last version was released less than a year ago. It has 2 open source maintainers collaborating on the project.
Did you know?
Socket for GitHub automatically highlights issues in each pull request and monitors the health of all your open source dependencies. Discover the contents of your packages and block harmful activity before you install or update your dependencies.
Security News
PyPI confirms no security flaws were exploited in the Ultralytics supply chain attack and highlights improvements for safer package publishing.
Security News
Research
The Socket Research Team breaks down a malicious wrapper package that uses obfuscation to harvest credentials and exfiltrate sensitive data.
Research
Security News
Attackers used a malicious npm package typosquatting a popular ESLint plugin to steal sensitive data, execute commands, and exploit developer systems.