helix.fhir.client.sdk
Fluent API to call the FHIR server that handles:
- Authentication to FHIR server
- Renewing access token when they expire
- Retry when there are transient errors
- Un-bundling the resources received from FHIR server
- Paging
- Streaming
- Logging
- Simulating a $graph call when the server does not support it
Usage
pip install helix.fhir.client.sdk
Documentation
https://icanbwell.github.io/helix.fhir.client.sdk/
Test Project using this
https://github.com/icanbwell/fhir-server-performance
Python Version Support
- 1.x supports python 3.7+
- 2.x supports python 3.10+
- 3.x supports python 3.12+
Asynchronous Support
When communicating with FHIR servers, a lot of time is spent waiting for the server to respond.
This is a good use case for using asynchronous programming.
This SDK supports asynchronous programming using the async and await keywords.
The return types are Python AsyncGenerators. Python makes it very easy to work with AsyncGenerators.
For example, if the SDK provides a function like this:
async def get_resources(self) -> AsyncGenerator[FhirGetResponse, None]:
...
You can iterate over the results as they become available:
response: Optional[FhirGetResponse]
async for response in client.get_resources():
print(response.resource)
Or you can get a list of responses (which will return AFTER all the responses are received:
responses: List[FhirGetResponse] = [response async for response in client.get_resources()]
Or you can aggregate the responses into one response (which will return AFTER all the responses are received:
response: Optional[FhirGetResponse] = await FhirGetResponse.from_async_generator(client.get_resources())
Data Streaming
For FHIR servers that support data streaming (e.g., b.well FHIR server), you can just set the use_data_streaming parameter to stream the data as it is received.
The data will be streamed in AsyncGenerators as described above.
Streaming $merge Responses
use_data_streaming also applies to $merge calls (merge_async, merge_resources_async, merge_bundle_async,
merge_bundle_uncompressed). When enabled, the SDK sends Accept: application/fhir+ndjson so a b.well FHIR server
that supports it (see merge.md) can stream the
$merge response back instead of building the whole response in memory first.
from helix_fhir_client_sdk.fhir_client import FhirClient
fhir_client = FhirClient().url(fhir_server_url).resource("Patient").use_data_streaming(True)
merge_response = await FhirMergeResponse.from_async_generator(
fhir_client.merge_async(json_data_list=[json.dumps(patient_resource)])
)
This only changes how the response is read; it does not change the request body. If you also want the request
body sent with chunked transfer encoding (independent of response streaming), opt in separately with
send_data_as_chunked(True).
Error responses (e.g. a 400 with an OperationOutcome) are unaffected by streaming - the SDK always returns the
full error body via response.error/response.responses[...]["issue"], whether or not use_data_streaming is
enabled.
Persistent Sessions (Connection Reuse)
By default, the SDK creates a new HTTP session for each request. For better performance (~4× faster), you can use persistent sessions to reuse connections across multiple requests.
Important: When you provide a custom session factory using use_http_session(), YOU are responsible
for managing the session lifecycle, including closing it when done. The SDK will NOT automatically close
user-provided sessions.
import aiohttp
from helix_fhir_client_sdk.fhir_client import FhirClient
# Create a persistent session for connection reuse
session = aiohttp.ClientSession()
try:
# Configure FhirClient to use persistent session
fhir_client = (
FhirClient()
.url("https://fhir.example.com")
.resource("Patient")
.use_http_session(lambda: session) # User provides session factory
)
# Multiple requests reuse the same connection (~4× performance boost)
response1 = await fhir_client.get_async()
response2 = await fhir_client.clone().resource("Observation").get_async()
finally:
# User must close the session when done
await session.close()
Session Lifecycle Rules:
- No custom factory (default): SDK creates and closes the session automatically
- Custom factory provided: User is responsible for closing the session
Storage Compression
The FHIR client SDK supports two types of compression:
- HTTP Compression (
compress): Compresses HTTP request body when sending data to the server. Default: enabled - In-Memory Storage (
storage_mode): Controls how FHIR resources are stored in memory. Default: raw (no compression)
Disabling HTTP Compression
HTTP compression (gzip) is enabled by default for request bodies. To disable it:
from helix_fhir_client_sdk.fhir_client import FhirClient
# Disable HTTP compression for requests
fhir_client = FhirClient().url("https://fhir.example.com").compress(False)
In-Memory Storage Modes
The SDK supports different storage modes for FHIR resources through the set_storage_mode() method.
By default, resources are stored as raw Python dictionaries (no compression).
from helix_fhir_client_sdk.fhir_client import FhirClient
from compressedfhir.utilities.compressed_dict.v1.compressed_dict_storage_mode import CompressedDictStorageMode
# Use raw storage (default) - no compression, resources stored as plain Python dicts
fhir_client = FhirClient().set_storage_mode(CompressedDictStorageMode(storage_type="raw"))
# Use msgpack storage - stores resources in msgpack format
fhir_client = FhirClient().set_storage_mode(CompressedDictStorageMode(storage_type="msgpack"))
# Use compressed msgpack storage - stores resources in compressed msgpack format
fhir_client = FhirClient().set_storage_mode(CompressedDictStorageMode(storage_type="compressed_msgpack"))
Available storage types:
raw: Default. Resources are stored as standard Python dictionaries (no compression)msgpack: Resources are serialized using MessagePack for efficient storagecompressed_msgpack: Resources are serialized using MessagePack and then compressed
Getting Raw Python Dictionaries
To completely bypass the compressedfhir library and get plain Python dictionaries:
# Returns plain Python dicts, not FhirResource objects
result = await fhir_client.get_raw_resources_async()
resources = result["_resources"] # list[dict[str, Any]]
1 maintainer
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Filter files by name, interpreter, ABI, and platform.
If you're not sure about the file name format, learn more about wheel file names.
Copy a direct link to the current filters
File details
Details for the file helix_fhir_client_sdk-5.0.5.tar.gz.
File metadata
- Download URL: helix_fhir_client_sdk-5.0.5.tar.gz
- Upload date:
- Size: 189.3 kB
- Tags: Source
- Uploaded using Trusted Publishing? No
- Uploaded via:
twine/7.0.0 CPython/3.12.14
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
27baa81601471ba4671ea546ca8be1ed4c6b3dd4a5ad43b9e9ad614bf5352a36
|
|
| MD5 |
7ac90778219f1fb767f148eaceb8b8ab
|
|
| BLAKE2b-256 |
ef414de3efd02e8d40a41ba426c2ce78e2a9d9cb1b123ffbd5f20b1bc7423946
|
File details
Details for the file helix_fhir_client_sdk-5.0.5-py3-none-any.whl.
File metadata
- Download URL: helix_fhir_client_sdk-5.0.5-py3-none-any.whl
- Upload date:
- Size: 321.3 kB
- Tags: Python 3
- Uploaded using Trusted Publishing? No
- Uploaded via:
twine/7.0.0 CPython/3.12.14
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
df7fb97ac278e7dcdc2664ab572b58e7d27d3adf3f99929a248ce372564969e3
|
|
| MD5 |
0896b651e511b1f58a10fec72423108b
|
|
| BLAKE2b-256 |
ef0fa310651f3861d58fe4ab8cd0454838c2f56bb85acef594516ca4b4d412af
|
