| Name | fhirpy JSON |
| Version |
1.4.2
JSON |
| download |
| home_page | |
| Summary | FHIR client for python |
| upload_time | 2023-09-04 11:16:36 |
| maintainer | |
| docs_url | None |
| author | |
| requires_python | >=3.8 |
| license | |
| keywords |
fhir
|
| VCS |
 |
| bugtrack_url |
|
| requirements |
No requirements were recorded.
|
| Travis-CI |
No Travis.
|
| coveralls test coverage |
No coveralls.
|
[](https://github.com/beda-software/fhir-py/actions/workflows/build.yaml)
[](https://codecov.io/gh/beda-software/fhir-py)
[](https://pypi.org/project/fhirpy)
[](https://www.python.org/downloads/release/python-380/)
# fhir-py
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](https://github.com/Aidbox/jupyter-course).
<!-- To regenerate table of contents: doctoc README.md --maxlevel=3 -->
<!-- START doctoc generated TOC please keep comment here to allow auto update -->
<!-- DON'T EDIT THIS SECTION, INSTEAD RE-RUN doctoc TO UPDATE -->
- [Getting started](#getting-started)
- [Async example](#async-example)
- [Searchset examples](#searchset-examples)
- [Chained parameters](#chained-parameters)
- [Reference](#reference)
- [Date](#date)
- [Modifiers](#modifiers)
- [Raw parameters](#raw-parameters)
- [Get resource by id](#get-resource-by-id)
- [Get exactly one resource](#get-exactly-one-resource)
- [Get first result](#get-first-result)
- [Get total count](#get-total-count)
- [Fetch one page](#fetch-one-page)
- [Fetch all resources on all pages](#fetch-all-resources-on-all-pages)
- [Page count (_count)](#page-count-_count)
- [Sort (_sort)](#sort-_sort)
- [Elements (_elements)](#elements-_elements)
- [Include](#include)
- [Modifier :iterate (or :recurse in some previous versions of FHIR)](#modifier-iterate-or-recurse-in-some-previous-versions-of-fhir)
- [Wild card (any search parameter of type=reference be included)](#wild-card-any-search-parameter-of-typereference-be-included)
- [Revinclude](#revinclude)
- [Wild card (any search parameter of type=reference be included)](#wild-card-any-search-parameter-of-typereference-be-included-1)
- [Conditional operations](#conditional-operations)
- [Conditional create](#conditional-create)
- [Conditional update](#conditional-update)
- [Conditional patch](#conditional-patch)
- [Resource and helper methods](#resource-and-helper-methods)
- [Validate resource using operation $validate](#validate-resource-using-operation-validate)
- [Accessing resource attributes](#accessing-resource-attributes)
- [get_by_path(path, default=None)](#get_by_pathpath-defaultnone)
- [set_by_path(obj, path, value)](#set_by_pathpath)
- [serialize()](#serialize)
- [Reference](#reference-1)
- [Main class structure](#main-class-structure)
- [Acync client (based on _aiohttp_) – AsyncFHIRClient](#acync-client-based-on-_aiohttp_--asyncfhirclient)
- [AsyncFHIRResource](#asyncfhirresource)
- [AsyncFHIRReference](#asyncfhirreference)
- [AsyncFHIRSearchSet](#asyncfhirsearchset)
- [Sync client (based on _requests_) – SyncFHIRClient](#sync-client-based-on-_requests_--syncfhirclient)
- [SyncFHIRResource](#syncfhirresource)
- [SyncFHIRReference](#syncfhirreference)
- [SyncFHIRSearchSet](#syncfhirsearchset)
- [Run integration tests](#run-integration-tests)
<!-- END doctoc generated TOC please keep comment here to allow auto update -->
# Getting started
## Async example
```Python
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())
```
## Searchset examples
```Python
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
```
### Chained parameters
```Python
patients.search(general_practitioner__Organization__name='Hospital')
# /Patient?general-practitioner:Organization.name=Hospital
```
```Python
patients.search(general_practitioner__name='Hospital')
# /Patient?general-practitioner.name=Hospital
```
### Reference
```Python
practitioner = client.resources('Practitioner').search(_id='john-smith').first()
patients.search(general_practitioner=practitioner)
# /Patient?general-practitioner=Practitioner/john-smith
```
### Date
```Python
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
```
### Modifiers
```Python
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
```
### Raw parameters
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
```Python
from fhirpy.base.searchset import Raw
patients = client.resources('Patient')
patients.search(Raw(**{'general-practitioner.name': 'Hospital'}))
# /Patient?general-practitioner.name=Hospital
```
## Get resource by id
Use reference to get resource by id
```Python
patient = await client.reference('Patient', '1').to_resource()
# /Patient/1
```
Or use FHIR search API with `.first()` or `.get()` as described below.
## Get exactly one resource
```Python
practitioners = client.resources('Practitioner')
try:
await practitioners.search(active=True, _id='id').get()
# /Practitioner?active=true&_id=id
except ResourceNotFound:
pass
except MultipleResourcesFound:
pass
```
## Get first result
```Python
await practitioners.search(name='Jack').first()
# /Practitioner?name=Jack&_count=1
await patients.sort('active', '-birthdate').first()
# /Patient?_sort=active,-birthdate&_count=1
```
## Get total count
```Python
await practitioners.search(active=True).count()
await patients.count()
```
## Fetch one page
```Python
await practitioners.fetch()
# /Practitioner
await patients.elements('name', 'telecom').fetch()
# /Patient?_elements=resourceType,name,id,telecom
```
## Fetch all resources on all pages
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.
```Python
# Returns a list of `Practitioner` resources
await practitioners.search(address_city='Krasnoyarsk').fetch_all()
await patients.fetch_all()
```
## Page count (_count)
```Python
# Get 100 resources
await practitioners.limit(100).fetch()
```
## Sort (_sort)
```Python
observations = client.resources('Observation')
observations.sort('status', '-date', 'category')
# /Observation?_sort=status,-date,category
```
## Elements (_elements)
```Python
# 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)
```
## Include
```Python
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
```
### Modifier :iterate (or :recurse in some previous versions of FHIR)
```Python
# 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
```
### Wild card (any search parameter of type=reference be included)
```Python
await client.resources('Encounter').include('*') \
.fetch_raw()
# /Encounter?_include=*
```
## Revinclude
```Python
await practitioners.revinclude('Group', 'member').fetch_raw()
# /Practitioner?_revinclude=Group:member
```
or
```Python
await practitioners.include('Group', 'member', reverse=True).fetch_raw()
# /Practitioner?_revinclude=Group:member
```
### Wild card (any search parameter of type=reference be included)
```Python
await client.resources('EpisodeOfCare').revinclude('*') \
.fetch_raw()
# /EpisodeOfCare?_revinclude=*
```
## Conditional operations
### Conditional create
[FHIR spec: Conditional create](https://build.fhir.org/http.html#ccreate)<br>
#### For resource
```Python
# 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")
```
#### For SearchSet
```Python
# 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'
```
### Conditional update
[FHIR spec: Conditional update](https://build.fhir.org/http.html#cond-update)<br>
```Python
# 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'
```
### Conditional patch
[FHIR spec: Conditional patch](https://build.fhir.org/http.html#cond-patch)<br>
```Python
# 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'
```
### Conditional delete
[FHIR spec: Conditional delete](https://build.fhir.org/http.html#cdelete)<br>
```Python
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)
```
# Resource and helper methods
## Validate resource using operation $validate
```Python
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
```
## Accessing resource attributes
```Python
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]
```
## Static type checking with [mypy](https://github.com/python/mypy) and [fhir-py-types](https://github.com/beda-software/fhir-py-types)
```Python
from fhir_py_types.generated.resources import Patient
patient: Patient = await client.resources('Patient').first()
# Works only with dictionary-like resource access
patient_family = patient['name'][0]['family1']
# 'TypedDict "HumanName" has no key "family1" note: Did you mean "family"?'
```
Check [fhir-py-types](https://github.com/beda-software/fhir-py-types) for more details on generating type definitions from FHIR `StructureDefintion`
## get_by_path(path, default=None)
```Python
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)
```
## set_by_path(obj, path, value)
```python
resource = {
"name": [{"given": ["Firstname"], "family": "Lastname"}],
}
set_by_path(resource, ["name", 0, "given", 0], "FirstnameUpdated")
# resource
# {"name": [{"given": ["FirstnameUpdated"], "family": "Lastname"}]}
```
## serialize()
```Python
# 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'}, ...}
```
# Reference
## Main class structure
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 |
## Acync client (based on _aiohttp_) – AsyncFHIRClient
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:
* .reference(resource_type, id, reference, **kwargs) - returns `AsyncFHIRReference` to the resource
* .resource(resource_type, **kwargs) - returns `AsyncFHIRResource` which described below
* .resources(resource_type) - returns `AsyncFHIRSearchSet`
* .execute(path, method='post', data=None, params=None) - returns a result of FHIR operation
### Aiohttp request parameters
Sometimes you need more control over the way http request is made and provide additional aiohttp [session's request](https://docs.aiohttp.org/en/stable/client_reference.html#aiohttp.ClientSession.request) parameters like `ssl`, `proxy`, `cookies`, `timeout` etc. It's possible by providing `aiohttp_config` dict for `AsyncFHIRClient`:
```Python
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.
### AsyncFHIRResource
provides:
* .serialize() - serializes resource
* .get_by_path(path, default=None) – gets the value at path of resource
* `async` .save(fields=[]) - creates or updates or patches (with fields=[...]) resource instance
* `async` .update() - overrides resource instance
* `async` .patch(**kwargs) - patches resource instance
* `async` .delete() - deletes resource instance
* `async` .refresh() - reloads resource from a server
* `async` .to_reference(**kwargs) - returns `AsyncFHIRReference` for this resource
* `async` .execute(operation, method='post', data=None, params=None) - returns a result of FHIR operation on the resource
### AsyncFHIRReference
provides:
* `async` .to_resource() - returns `AsyncFHIRResource` for this reference
* `async` .execute(operation, method='post', data=None, params=None) - returns a result of FHIR operation on the resource
### AsyncFHIRSearchSet
provides:
* .search(param=value)
* .limit(count)
* .sort(*args)
* .elements(*args, exclude=False)
* .include(resource_type, attr=None, recursive=False, iterate=False)
* .revinclude(resource_type, attr=None, recursive=False, iterate=False)
* .has(*args, **kwargs)
* `async` .fetch() - makes query to the server and returns a list of `Resource` filtered by resource type
* `async` .fetch_all() - makes query to the server and returns a full list of `Resource` filtered by resource type
* `async` .fetch_raw() - makes query to the server and returns a raw Bundle `Resource`
* `async` .first() - returns `Resource` or None
* `async` .get(id=None) - 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 SearchSet
* `async` .get_or_create(resource) - conditional create
* `async` .update(resource) - conditional update
* `async` .patch(resource) - conditional patch
## Sync client (based on _requests_) – SyncFHIRClient
Import 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:
* .reference(resource_type, id, reference, **kwargs) - returns `SyncFHIRReference` to the resource
* .resource(resource_type, **kwargs) - returns `SyncFHIRResource` which described below
* .resources(resource_type) - returns `SyncFHIRSearchSet`
### Requests request parameters
Pass `requests_config` parameter to `SyncFHIRClient` if you want to provide additional parameters for a [request](https://docs.python-requests.org/en/latest/api/#requests.request) like `verify`, `cert`, `timeout` etc.
```Python
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.
### SyncFHIRResource
The same as AsyncFHIRResource but with sync methods
### SyncFHIRReference
provides:
The same as AsyncFHIRReference but with sync methods
### SyncFHIRSearchSet
The same as AsyncFHIRSearchSet but with sync methods
# Run integration tests
(need some test FHIR server to run with, e.g. https://docs.aidbox.app/installation/setup-aidbox.dev)
1. Clone this repository:
`https://github.com/beda-software/fhir-py.git`
2. 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`)
3. Provide ENV variables `FHIR_SERVER_URL` and `FHIR_SERVER_AUTHORIZATION`, or edit tests/config.py
4. 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.
Raw data
{
"_id": null,
"home_page": "",
"name": "fhirpy",
"maintainer": "",
"docs_url": null,
"requires_python": ">=3.8",
"maintainer_email": "",
"keywords": "fhir",
"author": "",
"author_email": "\"beda.software\" <fhirpy@beda.software>",
"download_url": "https://files.pythonhosted.org/packages/4e/a3/4b59fce80d36711023f3d236c71a4e371a33c73ce6d3ba5d782ff82b8d3d/fhirpy-1.4.2.tar.gz",
"platform": null,
"description": "[](https://github.com/beda-software/fhir-py/actions/workflows/build.yaml)\n[](https://codecov.io/gh/beda-software/fhir-py)\n[](https://pypi.org/project/fhirpy)\n[](https://www.python.org/downloads/release/python-380/)\n\n# fhir-py\nasync/sync FHIR client for python3.\nThis package provides an API for CRUD operations over FHIR resources\n\n```pip install fhirpy```\n\nor to install the latest dev version:\n\n```pip install git+https://github.com/beda-software/fhir-py.git```\n\nYou can test this library by interactive FHIR course in the repository [Aidbox/jupyter-course](https://github.com/Aidbox/jupyter-course).\n\n<!-- To regenerate table of contents: doctoc README.md --maxlevel=3 -->\n<!-- START doctoc generated TOC please keep comment here to allow auto update -->\n<!-- DON'T EDIT THIS SECTION, INSTEAD RE-RUN doctoc TO UPDATE -->\n\n- [Getting started](#getting-started)\n - [Async example](#async-example)\n - [Searchset examples](#searchset-examples)\n - [Chained parameters](#chained-parameters)\n - [Reference](#reference)\n - [Date](#date)\n - [Modifiers](#modifiers)\n - [Raw parameters](#raw-parameters)\n - [Get resource by id](#get-resource-by-id)\n - [Get exactly one resource](#get-exactly-one-resource)\n - [Get first result](#get-first-result)\n - [Get total count](#get-total-count)\n - [Fetch one page](#fetch-one-page)\n - [Fetch all resources on all pages](#fetch-all-resources-on-all-pages)\n - [Page count (_count)](#page-count-_count)\n - [Sort (_sort)](#sort-_sort)\n - [Elements (_elements)](#elements-_elements)\n - [Include](#include)\n - [Modifier :iterate (or :recurse in some previous versions of FHIR)](#modifier-iterate-or-recurse-in-some-previous-versions-of-fhir)\n - [Wild card (any search parameter of type=reference be included)](#wild-card-any-search-parameter-of-typereference-be-included)\n - [Revinclude](#revinclude)\n - [Wild card (any search parameter of type=reference be included)](#wild-card-any-search-parameter-of-typereference-be-included-1)\n - [Conditional operations](#conditional-operations)\n - [Conditional create](#conditional-create)\n - [Conditional update](#conditional-update)\n - [Conditional patch](#conditional-patch)\n- [Resource and helper methods](#resource-and-helper-methods)\n - [Validate resource using operation $validate](#validate-resource-using-operation-validate)\n - [Accessing resource attributes](#accessing-resource-attributes)\n - [get_by_path(path, default=None)](#get_by_pathpath-defaultnone)\n - [set_by_path(obj, path, value)](#set_by_pathpath)\n - [serialize()](#serialize)\n- [Reference](#reference-1)\n - [Main class structure](#main-class-structure)\n - [Acync client (based on _aiohttp_) \u2013 AsyncFHIRClient](#acync-client-based-on-_aiohttp_--asyncfhirclient)\n - [AsyncFHIRResource](#asyncfhirresource)\n - [AsyncFHIRReference](#asyncfhirreference)\n - [AsyncFHIRSearchSet](#asyncfhirsearchset)\n - [Sync client (based on _requests_) \u2013 SyncFHIRClient](#sync-client-based-on-_requests_--syncfhirclient)\n - [SyncFHIRResource](#syncfhirresource)\n - [SyncFHIRReference](#syncfhirreference)\n - [SyncFHIRSearchSet](#syncfhirsearchset)\n- [Run integration tests](#run-integration-tests)\n\n<!-- END doctoc generated TOC please keep comment here to allow auto update -->\n\n# Getting started\n## Async example\n```Python\nimport asyncio\nfrom fhirpy import AsyncFHIRClient\n\n\nasync def main():\n # Create an instance\n client = AsyncFHIRClient(\n 'http://fhir-server/',\n authorization='Bearer TOKEN',\n )\n\n # Search for patients\n resources = client.resources('Patient') # Return lazy search set\n resources = resources.search(name='John').limit(10).sort('name')\n patients = await resources.fetch() # Returns list of AsyncFHIRResource\n\n # Create Organization resource\n organization = client.resource(\n 'Organization',\n name='beda.software',\n active=False\n )\n await organization.save()\n\n # Update (PATCH) organization. Resource support accessing its elements\n # both as attribute and as a dictionary keys\n if organization['active'] is False:\n organization.active = True\n await organization.save(fields=['active'])\n # `await organization.patch(active=True)` would do the same PATCH operation\n\n # Get patient resource by reference and delete\n patient_ref = client.reference('Patient', 'new_patient')\n # Get resource from this reference\n # (throw ResourceNotFound if no resource was found)\n patient_res = await patient_ref.to_resource()\n await patient_res.delete()\n\n # Iterate over search set\n org_resources = client.resources('Organization')\n # Lazy loading resources page by page with page count = 100\n async for org_resource in org_resources.limit(100):\n print(org_resource.serialize())\n\n\nif __name__ == '__main__':\n loop = asyncio.get_event_loop()\n loop.run_until_complete(main())\n```\n\n## Searchset examples\n```Python\npatients = client.resources('Patient')\n\npatients.search(birthdate__gt='1944', birthdate__lt='1964')\n# /Patient?birthdate=gt1944&birthdate=lt1964\n\npatients.search(name__contains='John')\n# /Patient?name:contains=John\n\npatients.search(name=['John', 'Rivera'])\n# /Patient?name=John&name=Rivera\n\npatients.search(name='John,Eva')\n# /Patient?name=John,Eva\n\npatients.search(family__exact='Moore')\n# /Patient?family:exact=Moore\n\npatients.search(address_state='TX')\n# /Patient?address-state=TX\n\npatients.search(active=True, _id='id')\n# /Patient?active=true&_id=id\n\npatients.search(gender__not=['male', 'female'])\n# /Patient?gender:not=male&gender:not=female\n```\n\n### Chained parameters\n```Python\npatients.search(general_practitioner__Organization__name='Hospital')\n# /Patient?general-practitioner:Organization.name=Hospital\n```\n\n```Python\npatients.search(general_practitioner__name='Hospital')\n# /Patient?general-practitioner.name=Hospital\n```\n\n### Reference\n```Python\npractitioner = client.resources('Practitioner').search(_id='john-smith').first()\npatients.search(general_practitioner=practitioner)\n# /Patient?general-practitioner=Practitioner/john-smith\n```\n\n### Date\n```Python\nimport pytz\nimport datetime\n\n\npatients.search(birthdate__lt=datetime.datetime.now(pytz.utc))\n# /Patient?birthdate=lt2019-11-19T20:16:08Z\n\npatients.search(birthdate__gt=datetime.datetime(2013, 10, 27, tzinfo=pytz.utc))\n# /Patient?birthdate=gt2013-10-27T00:00:00Z\n```\n\n### Modifiers\n```Python\nconditions = client.resources('Condition')\n\nconditions.search(code__text='headache')\n# /Condition?code:text=headache\n\nconditions.search(code__in='http://acme.org/fhir/ValueSet/cardiac-conditions')\n# /Condition?code:in=http://acme.org/fhir/ValueSet/cardiac-conditions\n\nconditions.search(code__not_in='http://acme.org/fhir/ValueSet/cardiac-conditions')\n# /Condition?code:not-in=http://acme.org/fhir/ValueSet/cardiac-conditions\n\nconditions.search(code__below='126851005')\n# /Condition?code:below=126851005\n\nconditions.search(code__above='126851005')\n# /Condition?code:above=126851005\n```\n\n### Raw parameters\nSometimes you can find that fhir-py does not implement some search parameters from the FHIR specification. \nIn this case, you can use `Raw()` wrapper without any transformations\n\n```Python\nfrom fhirpy.base.searchset import Raw\n\npatients = client.resources('Patient')\npatients.search(Raw(**{'general-practitioner.name': 'Hospital'}))\n# /Patient?general-practitioner.name=Hospital\n```\n\n## Get resource by id\nUse reference to get resource by id\n```Python\npatient = await client.reference('Patient', '1').to_resource()\n# /Patient/1\n```\n\nOr use FHIR search API with `.first()` or `.get()` as described below.\n\n## Get exactly one resource\n```Python\npractitioners = client.resources('Practitioner')\n\ntry:\n await practitioners.search(active=True, _id='id').get()\n # /Practitioner?active=true&_id=id\nexcept ResourceNotFound:\n pass\nexcept MultipleResourcesFound:\n pass\n```\n\n## Get first result\n```Python\nawait practitioners.search(name='Jack').first()\n# /Practitioner?name=Jack&_count=1\n\nawait patients.sort('active', '-birthdate').first()\n# /Patient?_sort=active,-birthdate&_count=1\n```\n\n## Get total count\n```Python\nawait practitioners.search(active=True).count()\n\nawait patients.count()\n```\n\n## Fetch one page\n```Python\nawait practitioners.fetch()\n# /Practitioner\n\nawait patients.elements('name', 'telecom').fetch()\n# /Patient?_elements=resourceType,name,id,telecom\n```\n\n## Fetch all resources on all pages\nKeep 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.\n```Python\n# Returns a list of `Practitioner` resources\nawait practitioners.search(address_city='Krasnoyarsk').fetch_all()\n\nawait patients.fetch_all()\n```\n\n## Page count (_count)\n```Python\n# Get 100 resources\nawait practitioners.limit(100).fetch()\n```\n\n## Sort (_sort)\n```Python\nobservations = client.resources('Observation')\n\nobservations.sort('status', '-date', 'category')\n# /Observation?_sort=status,-date,category\n```\n\n## Elements (_elements)\n```Python\n# Get only specified set of elements for each resource\npatients.elements('identifier', 'active', 'link')\n# /Patient?_elements=identifier,active,link\n\n# Get all elements except specified set\npractitioners.elements('address', 'telecom', exclude=True)\n```\n\n## Include\n```Python\nresult = await client.resources('EpisodeOfCare') \\\n .include('EpisodeOfCare', 'patient').fetch_raw()\n# /EpisodeOfCare?_include=EpisodeOfCare:patient\nfor entry in result.entry:\n print(entry.resource)\n\nawait client.resources('MedicationRequest') \\\n .include('MedicationRequest', 'patient', target_resource_type='Patient') \\\n .fetch_raw()\n# /MedicationRequest?_include=MedicationRequest:patient:Patient\n```\n### Modifier :iterate (or :recurse in some previous versions of FHIR)\n```Python\n# For FHIR version >= 3.5 we can also use modifier :iterate\nawait client.resources('MedicationRequest') \\\n .include('MedicationDispense', 'prescription') \\\n .include('MedicationRequest', 'performer', iterate=True) \\\n .fetch_raw()\n# /MedicationRequest?_include=MedicationDispense:prescription\n# &_include:iterate=MedicationRequest:performer\n\n# For FHIR version 3.0-3.3 use modifier :recurse\nawait client.resources('MedicationDispense') \\\n .include('MedicationRequest', 'prescriber', recursive=True) \\\n .fetch_raw()\n# /MedicationDispense?_include:recurse=MedicationRequest:prescriber\n```\n### Wild card (any search parameter of type=reference be included)\n```Python\nawait client.resources('Encounter').include('*') \\\n .fetch_raw()\n# /Encounter?_include=*\n```\n\n## Revinclude\n```Python\nawait practitioners.revinclude('Group', 'member').fetch_raw()\n# /Practitioner?_revinclude=Group:member\n```\nor\n```Python\nawait practitioners.include('Group', 'member', reverse=True).fetch_raw()\n# /Practitioner?_revinclude=Group:member\n```\n\n### Wild card (any search parameter of type=reference be included)\n```Python\nawait client.resources('EpisodeOfCare').revinclude('*') \\\n .fetch_raw()\n# /EpisodeOfCare?_revinclude=*\n```\n\n## Conditional operations\n### Conditional create\n[FHIR spec: Conditional create](https://build.fhir.org/http.html#ccreate)<br>\n#### For resource\n```Python\n# resource.create(search_params)\n# executes POST /Patient?identifier=fhirpy\n\npatient = client.resource(\"Patient\",\n identifier=[{\"system\": \"http://example.com/env\", \"value\": \"fhirpy\"}],\n name=[{\"text\": \"Mr. Smith\"}],\n)\nawait patient.create(identifier=\"other\")\n```\n#### For SearchSet\n```Python\n# searchset.get_or_create(resource)\n# executes POST /Patient?identifier=fhirpy\n\npatient, created = await client.resources(\"Patient\").search(identifier=\"fhirpy\").get_or_create(patient_to_save)\n\n# no match -> created is True\n# one match -> created is False, return existing resource\n# multiple matches -> 412 'MultipleResourcesFound'\n```\n### Conditional update\n[FHIR spec: Conditional update](https://build.fhir.org/http.html#cond-update)<br>\n```Python\n# resource, created: bool = searchset.patch(resource)\n# executes PUT /Patient?identifier=fhirpy\n\npatient_to_update = client.resource(\"Patient\", \n identifier=[{\"system\": \"http://example.com/env\", \"value\": \"fhirpy\"}],\n active=False)\nnew_patient, created = await client.resources(\"Patient\").search(identifier=\"fhirpy\").update(patient_to_update)\n\n# no match -> created is True\n# one match -> created is False, the matched resource is overwritten\n# multiple matches -> 412 'MultipleResourcesFound'\n```\n### Conditional patch\n[FHIR spec: Conditional patch](https://build.fhir.org/http.html#cond-patch)<br>\n```Python\n# patched_resource = searchset.patch(resource)\n# executes PATCH /Patient?identifier=fhirpy\n\npatient_to_patch = client.resource(\"Patient\", \n identifier=[{\"system\": \"http://example.com/env\", \"value\": \"fhirpy\"}], \n name=[{\"text\": \"Mr. Smith\"}])\npatched_patient = await client.resources(\"Patient\").search(identifier=\"fhirpy\").patch(patient_to_patch)\n\n# no match -> 404 'ResourceNotFound'\n# multiple matches -> 412 'MultipleResourcesFound'\n```\n\n### Conditional delete\n[FHIR spec: Conditional delete](https://build.fhir.org/http.html#cdelete)<br>\n```Python\nresponse_data, status_code = await self.client.resources(\"Patient\").search(identifier=\"abc\").delete()\n\n# no match -> status_code = 204 'No Content'\n# one match -> status_code = 200 'OK'\n# multiple matches -> status_code = 412 'MultipleResourcesFound' (implementation specific)\n```\n\n# Resource and helper methods\n\n## Validate resource using operation $validate\n```Python\ntry:\n await client.resource('Patient', birthDate='date', custom_prop='123', telecom=True) \\\n .is_valid(raise_exception=True)\nexcept OperationOutcome as e:\n print('Error: {}'.format(e))\n\npatient = client.resource('Patient', birthDate='1998-01-01')\nif (await patient.is_valid()):\n pass\n```\n\n## Accessing resource attributes\n```Python\npatient = await client.resources('Patient').first()\n\n# Work with the resource as a dictionary\npatient_family = patient['name'][0]['family']\n\n# Or access value by an attribute\npatient_given_name = patient.name[0].given[0]\n```\n\n## Static type checking with [mypy](https://github.com/python/mypy) and [fhir-py-types](https://github.com/beda-software/fhir-py-types)\n```Python\nfrom fhir_py_types.generated.resources import Patient\n\npatient: Patient = await client.resources('Patient').first()\n\n# Works only with dictionary-like resource access\npatient_family = patient['name'][0]['family1']\n# 'TypedDict \"HumanName\" has no key \"family1\" note: Did you mean \"family\"?'\n```\nCheck [fhir-py-types](https://github.com/beda-software/fhir-py-types) for more details on generating type definitions from FHIR `StructureDefintion`\n\n## get_by_path(path, default=None)\n```Python\npatient_postal = patient.get_by_path(['resource', 'address', 0, 'postalCode'])\n\n# get_by_path can be also used on any nested attribute\npatient_name = patient.name[0]\npatient_fullname = '{} {}'.format(\n patient_name.get_by_path(['given', 0]),\n patient_name.get_by_path(['family'])\n)\n\n# Get identifier value by specified system or empty string\nuid = patient.get_by_path([\n 'resource', 'identifier',\n {'system':'http://example.com/identifier/uid'},\n 'value'\n ], '')\n\n# Get base value amount or 0\ninvoice = await client.resources('Invoice').first()\nbase_value = invoice.get_by_path([\n 'totalPriceComponent',\n {'type': 'base'},\n 'amount', 'value'], 0)\n```\n\n## set_by_path(obj, path, value)\n```python\nresource = {\n \"name\": [{\"given\": [\"Firstname\"], \"family\": \"Lastname\"}],\n}\n\nset_by_path(resource, [\"name\", 0, \"given\", 0], \"FirstnameUpdated\")\n\n# resource\n# {\"name\": [{\"given\": [\"FirstnameUpdated\"], \"family\": \"Lastname\"}]}\n```\n\n## serialize()\n```Python\n# Returns resources as dict\npatient = await client.reference('Patient', '1').to_resource()\npatient.serialize()\n# Or \nawait client.reference('Patient', '1').to_resource().serialize()\n# {'resourceType': 'Patient', 'id': '1', 'meta': {'versionId': '1', 'lastUpdated': '2021-11-13T11:50:24.685719Z'}, ...}\n```\n\n# Reference\n\n## Main class structure\nBoth async and sync clients have identical sets of classes and methods.\n\n| | Sync | Async |\n| ------------- | ------------------- | -------------------- |\n| Client | SyncFHIRClient | AsyncFHIRClient |\n| SearchSet | SyncFHIRSearchSet | AsyncFHIRSearchSet |\n| Resource | SyncFHIRResource | AsyncFHIRResource |\n| Reference | SyncFHIRReference | AsyncFHIRReference |\n\n\n## Acync client (based on _aiohttp_) \u2013 AsyncFHIRClient\nImport library:\n\n`from fhirpy import AsyncFHIRClient`\n\nTo create AsyncFHIRClient instance use:\n\n`AsyncFHIRClient(url, authorization='', extra_headers={})`\n\nReturns an instance of the connection to the server which provides:\n* .reference(resource_type, id, reference, **kwargs) - returns `AsyncFHIRReference` to the resource\n* .resource(resource_type, **kwargs) - returns `AsyncFHIRResource` which described below\n* .resources(resource_type) - returns `AsyncFHIRSearchSet`\n* .execute(path, method='post', data=None, params=None) - returns a result of FHIR operation\n\n### Aiohttp request parameters\nSometimes you need more control over the way http request is made and provide additional aiohttp [session's request](https://docs.aiohttp.org/en/stable/client_reference.html#aiohttp.ClientSession.request) parameters like `ssl`, `proxy`, `cookies`, `timeout` etc. It's possible by providing `aiohttp_config` dict for `AsyncFHIRClient`:\n```Python\nclient = AsyncFHIRClient(\n FHIR_SERVER_URL,\n aiohttp_config={\n \"ssl\": ssl.create_default_context(),\n \"timeout\": aiohttp.ClientTimeout(total=100),\n }\n)\n```\n\nBe 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. \n\n### AsyncFHIRResource\n\nprovides:\n* .serialize() - serializes resource\n* .get_by_path(path, default=None) \u2013 gets the value at path of resource\n* `async` .save(fields=[]) - creates or updates or patches (with fields=[...]) resource instance\n* `async` .update() - overrides resource instance\n* `async` .patch(**kwargs) - patches resource instance\n* `async` .delete() - deletes resource instance\n* `async` .refresh() - reloads resource from a server\n* `async` .to_reference(**kwargs) - returns `AsyncFHIRReference` for this resource\n* `async` .execute(operation, method='post', data=None, params=None) - returns a result of FHIR operation on the resource\n\n### AsyncFHIRReference\n\nprovides:\n* `async` .to_resource() - returns `AsyncFHIRResource` for this reference\n* `async` .execute(operation, method='post', data=None, params=None) - returns a result of FHIR operation on the resource\n\n### AsyncFHIRSearchSet\n\nprovides:\n* .search(param=value)\n* .limit(count)\n* .sort(*args)\n* .elements(*args, exclude=False)\n* .include(resource_type, attr=None, recursive=False, iterate=False)\n* .revinclude(resource_type, attr=None, recursive=False, iterate=False)\n* .has(*args, **kwargs)\n* `async` .fetch() - makes query to the server and returns a list of `Resource` filtered by resource type\n* `async` .fetch_all() - makes query to the server and returns a full list of `Resource` filtered by resource type\n* `async` .fetch_raw() - makes query to the server and returns a raw Bundle `Resource`\n* `async` .first() - returns `Resource` or None\n* `async` .get(id=None) - returns `Resource` or raises `ResourceNotFound` when no resource found or MultipleResourcesFound when more than one resource found (parameter 'id' is deprecated)\n* `async` .count() - makes query to the server and returns the total number of resources that match the SearchSet\n* `async` .get_or_create(resource) - conditional create\n* `async` .update(resource) - conditional update\n* `async` .patch(resource) - conditional patch\n\n\n## Sync client (based on _requests_) \u2013 SyncFHIRClient\nImport library:\n\n`from fhirpy import SyncFHIRClient`\n\nTo create SyncFHIRClient instance use:\n\n`SyncFHIRClient(url, authorization='', extra_headers={})`\n\n\nReturns an instance of the connection to the server which provides:\n* .reference(resource_type, id, reference, **kwargs) - returns `SyncFHIRReference` to the resource\n* .resource(resource_type, **kwargs) - returns `SyncFHIRResource` which described below\n* .resources(resource_type) - returns `SyncFHIRSearchSet`\n\n### Requests request parameters\nPass `requests_config` parameter to `SyncFHIRClient` if you want to provide additional parameters for a [request](https://docs.python-requests.org/en/latest/api/#requests.request) like `verify`, `cert`, `timeout` etc.\n```Python\nclient = SyncFHIRClient(\n FHIR_SERVER_URL,\n requests_config={\n \"verify\": False,\n \"allow_redirects\": True,\n \"timeout\": 60,\n }\n)\n```\n\nBe 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. \n\n### SyncFHIRResource\n\nThe same as AsyncFHIRResource but with sync methods\n\n### SyncFHIRReference\n\nprovides:\nThe same as AsyncFHIRReference but with sync methods\n\n### SyncFHIRSearchSet\n\nThe same as AsyncFHIRSearchSet but with sync methods\n\n\n# Run integration tests\n(need some test FHIR server to run with, e.g. https://docs.aidbox.app/installation/setup-aidbox.dev)\n1. Clone this repository:\n`https://github.com/beda-software/fhir-py.git`\n\n2. Go to fhir-py folder and install dev dependencies:\n```\ncd fhir-py\npip install -r requirements.txt\n```\n\nIf 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`)\n\n3. Provide ENV variables `FHIR_SERVER_URL` and `FHIR_SERVER_AUTHORIZATION`, or edit tests/config.py\n\n4. Run `pytest`\n\nIf 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.\n\n",
"bugtrack_url": null,
"license": "",
"summary": "FHIR client for python",
"version": "1.4.2",
"project_urls": {
"Documentation": "https://github.com/beda-software/fhir-py#readme",
"Homepage": "https://github.com/beda-software/fhir-py",
"Source": "https://github.com/beda-software/fhir-py.git"
},
"split_keywords": [
"fhir"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "676f2a61fcc29973e1e91f6f4ce668f3eb3a440c28ea99a9d122499e3a145038",
"md5": "10af4919fbabd247dde493fd9b98f8af",
"sha256": "cc41da118a0248f60021ccc59f2ce7cd5af59b7926a60928a88563770bf1f59f"
},
"downloads": -1,
"filename": "fhirpy-1.4.2-py3-none-any.whl",
"has_sig": false,
"md5_digest": "10af4919fbabd247dde493fd9b98f8af",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.8",
"size": 20579,
"upload_time": "2023-09-04T11:16:35",
"upload_time_iso_8601": "2023-09-04T11:16:35.525172Z",
"url": "https://files.pythonhosted.org/packages/67/6f/2a61fcc29973e1e91f6f4ce668f3eb3a440c28ea99a9d122499e3a145038/fhirpy-1.4.2-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "4ea34b59fce80d36711023f3d236c71a4e371a33c73ce6d3ba5d782ff82b8d3d",
"md5": "2e6166c617453cca45448f148b399982",
"sha256": "30712451405047426582bc22598f395913b53f3b3cda8426a390692239030fc6"
},
"downloads": -1,
"filename": "fhirpy-1.4.2.tar.gz",
"has_sig": false,
"md5_digest": "2e6166c617453cca45448f148b399982",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.8",
"size": 23889,
"upload_time": "2023-09-04T11:16:36",
"upload_time_iso_8601": "2023-09-04T11:16:36.895493Z",
"url": "https://files.pythonhosted.org/packages/4e/a3/4b59fce80d36711023f3d236c71a4e371a33c73ce6d3ba5d782ff82b8d3d/fhirpy-1.4.2.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2023-09-04 11:16:36",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "beda-software",
"github_project": "fhir-py#readme",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "fhirpy"
}
