API docs
FhirServer
capabilities: CapabilityStatement
property
readonly
Get the capabilities statement for the server
rest_resources: List[str]
property
readonly
Get the list of resources available on the server
headers
property
readonly
auth
property
readonly
__init__(self, api_address, username=None, password=None, token=None, client_id=None, client_secret=None, oidc_provider_url=None, auth=None, headers=None, proxies=None, timeout=None, fhir_server_type='hapi', retry_status_codes=None, retryable_methods=None, max_atttempts=5, max_backoff_wait=60, backoff_factor=0.1, jitter_ratio=0.1, respect_retry_after_header=True)
special
Initialize a FHIR server connection
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
api_address |
str |
the base endpoint of the fhir server REST API |
required |
username |
Optional[str] |
username for basic auth |
None |
password |
Optional[str] |
password for basic auth |
None |
token |
Optional[str] |
token for static token auth |
None |
client_id |
Optional[str] |
client id for oauth2 |
None |
client_secret |
Optional[str] |
client secret for oauth2 |
None |
oidc_provider_url |
Optional[str] |
provider url for oauth2 |
None |
auth |
Optional[httpx.Auth] |
optional auth object to authenticate against a server |
None |
headers |
Optional[dict] |
optional additional headers to be added to the session |
None |
proxies |
Union[dict, str] |
optional proxies to be added to the session |
None |
timeout |
Optional[int] |
optional timeout for the session default is None |
None |
fhir_server_type |
str |
type of fhir server (hapi, blaze, etc.) |
'hapi' |
retry_status_codes |
Optional[Iterable[int]] |
optional list of status codes to retry on |
None |
max_atttempts |
int |
optional number of times to retry |
5 |
retry_wait |
optional number of seconds to wait between retries |
required |
from_env(no_auth=False)
classmethod
query(self, resource=None, query_string=None, query_parameters=None, output_format='json')
Initialize a FHIR query against the server with the given resource, query parameters or query string
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
output_format |
str |
the output format to request from the fhir server (json or xml) defaults to json |
'json' |
query_string |
str |
preformatted query string to execute against the servers REST API |
None |
query_parameters |
FhirQueryParameters |
optionally pass in a query parameters object to use for the query |
None |
resource |
Union[fhir.resources.resource.Resource, fhir.resources.core.fhirabstractmodel.FHIRAbstractModel, str] |
the FHIR resource to query from the server |
None |
Returns:
| Type | Description |
|---|---|
FhirQuerySync |
a FhirQuerySync object that can be further modified with filters and conditions before being executed against the server |
query_async(self, resource=None, query_string=None, query_parameters=None, output_format='json')
Initialize an asynchronous FHIR query against the server with the given resource, query parameters or query string
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
output_format |
str |
the output format to request from the fhir server (json or xml) defaults to json |
'json' |
query_string |
str |
preformatted query string to execute against the servers REST API |
None |
query_parameters |
FhirQueryParameters |
optionally pass in a query parameters object to use for the query |
None |
resource |
Union[fhir.resources.resource.Resource, fhir.resources.core.fhirabstractmodel.FHIRAbstractModel, str] |
the FHIR resource to query from the server |
None |
Returns:
| Type | Description |
|---|---|
FhirQueryAsync |
a FhirQueryAsync object that can be further modified with filters and conditions before being executed against the server |
raw_query(self, query_string, output_format='json')
Execute a raw query string against the server
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
query_string |
str |
required | |
output_format |
str |
'json' |
Returns:
| Type | Description |
|---|---|
FhirQuerySync |
a FhirQuerySync object that can be further modified with filters and conditions before being executed against the server |
raw_query_async(self, query_string, output_format='json')
Asynchronously Execute a raw query string against the server
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
query_string |
str |
query string defining the query to execute |
required |
output_format |
str |
the output format to request from the fhir server (json or xml) defaults to json |
'json' |
Returns:
| Type | Description |
|---|---|
FhirQueryAsync |
a FhirQueryAsync object that can be further modified with filters and conditions before being executed against the server |
get(self, reference)
Get a resource from the server specified by the given reference {ResourceType}/{id}
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
reference |
Union[str, fhir.resources.reference.Reference] |
reference to the resource, either a Reference object or a string of the form {ResourceType}/{id} |
required |
Returns:
| Type | Description |
|---|---|
FHIRAbstractModel |
the resource from the server specified by the reference |
get_async(self, reference)
async
Asynchronously get a resource from the server specified by the given reference {ResourceType}/{id}
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
reference |
Union[str, fhir.resources.reference.Reference] |
reference to the resource, either a Reference object or a string of the form {ResourceType}/{id} |
required |
Returns:
| Type | Description |
|---|---|
FHIRAbstractModel |
the resource from the server specified by the reference |
get_many(self, references)
Get a list of resources from the server specified by the given references
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
references |
List[Union[str, fhir.resources.reference.Reference]] |
list of references to the resources, either a Reference object or a string of the form
|
required |
Returns:
| Type | Description |
|---|---|
List[fhir.resources.core.fhirabstractmodel.FHIRAbstractModel] |
list of resources corresponding to the references |
get_many_async(self, references)
async
Asynchronously get a list of resources from the server specified by the given references
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
references |
List[Union[str, fhir.resources.reference.Reference]] |
list of references to the resources, either a Reference object or a string of the form
|
required |
Returns:
| Type | Description |
|---|---|
List[fhir.resources.core.fhirabstractmodel.FHIRAbstractModel] |
list of resources corresponding to the references |
add(self, resource)
Upload a resource to the server
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
resource |
Union[fhir.resources.resource.Resource, dict] |
dictionary containing the resource or FHIR resource object to be uploaded to the server |
required |
Returns:
| Type | Description |
|---|---|
ResourceCreateResponse |
CreateResponse containing the server response and the resource that was uploaded |
add_async(self, resource)
async
Asynchronously upload a resource to the server
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
resource |
Union[fhir.resources.resource.Resource, dict] |
dictionary containing the resource or FHIR resource object to be uploaded to the server |
required |
Returns:
| Type | Description |
|---|---|
ResourceCreateResponse |
CreateResponse containing the server response and the resource that was uploaded |
add_all(self, resources, batch_size=1000, display=True)
Upload a list of resources to the server, after packaging them into a bundle
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
resources |
List[Union[fhir.resources.resource.Resource, fhir.resources.core.fhirabstractmodel.FHIRAbstractModel, dict]] |
list of resources to upload to the server, either dictionary or FHIR resource objects |
required |
batch_size |
int |
maximum number of resources to upload in one bundle |
1000 |
display |
bool |
whether to display a progress bar when the upload is batched |
True |
Returns:
| Type | Description |
|---|---|
BundleCreateResponse |
Bundle create response from the fhir server |
add_all_async(self, resources, batch_size=5000, display=True)
async
Asynchronously upload a list of resources to the server, after packaging them into a bundle
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
resources |
List[Union[fhir.resources.resource.Resource, fhir.resources.core.fhirabstractmodel.FHIRAbstractModel, dict]] |
list of resources to upload to the server, either dictionary or FHIR resource objects |
required |
batch_size |
int |
maximum number of resources to upload in one bundle |
5000 |
display |
bool |
whether to display a progress bar when the upload is batched |
True |
Returns: Bundle create response from the fhir server
add_bundle(self, bundle, validate=True)
Upload a bundle to the server :param bundle: str, dict or Bundle object to upload to the server :param validate: whether to validate the entries in the bundle :return: BundleCreateResponse from the fhir server containing the server assigned ids of the resources in the bundle
add_bundle_async(self, bundle, validate_entries=True)
async
Asynchronously upload a bundle to the server :param bundle: str, dict or Bundle object to upload to the server :param validate_entries: whether to validate the entries in the bundle :return: BundleCreateResponse from the fhir server containing the server assigned ids of the resources in the bundle
update(self, resources)
Update a list of resources that exist on the server
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
resources |
List[Union[fhir.resources.fhirresourcemodel.FHIRResourceModel, dict]] |
List of updated resources coming to send to the server |
required |
Returns: Bundle update response from the fhir server
update_async(self, resources)
async
delete(self, resources=None, references=None, query=None)
Delete resources from the server. Either resources, references or a query must be specified.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
resources |
List[Union[fhir.resources.fhirresourcemodel.FHIRResourceModel, dict]] |
Resources coming from the server containing an id to delete |
None |
references |
List[Union[str, fhir.resources.reference.Reference]] |
references {Resource}/{id} to delete |
None |
query |
FhirQuerySync |
query to use to find resources to delete |
None |
Returns:
| Type | Description |
|---|---|
None |
None |
delete_async(self, resources=None, references=None, query=None)
async
Asynchronously delete resources from the server. Either resources, references or a query must be specified.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
resources |
List[Union[fhir.resources.fhirresourcemodel.FHIRResourceModel, dict]] |
Resources coming from the server containing an id to delete |
None |
references |
List[Union[str, fhir.resources.reference.Reference]] |
references {Resource}/{id} to delete |
None |
query |
FhirQueryAsync |
query to use to find resources to delete |
None |
Returns:
| Type | Description |
|---|---|
None |
None |
transfer(self, target_server, query=None, resources=None, get_missing=True, record_linkage=True, display=False)
Transfer resources from this server to another server while using server assigned ids and keeping referential integrity.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
target_server |
FhirServer |
FhirServer to transfer to |
required |
query |
FhirQuerySync |
FhirQuerySync to use to find resources to transfer |
None |
resources |
List[Union[fhir.resources.resource.Resource, fhir.resources.core.fhirabstractmodel.FHIRAbstractModel]] |
list of resources to transfer |
None |
get_missing |
bool |
whether to get missing references from the source server |
True |
record_linkage |
bool |
whether to record the linkage between the source and target server |
True |
display |
bool |
whether to display the progress bar |
False |
Returns:
| Type | Description |
|---|---|
TransferResponse |
Transfer response for the transfer of the query result to the target server |
summary(self, display=True)
Create a summary for the server. Contains resource counts for all resources available on the server.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
display |
bool |
whether to display a progress bar |
True |
Returns:
| Type | Description |
|---|---|
ServerSummary |
ServerSummary containing resource counts for all resources available on the server |
summary_async(self, display=True)
async
Asynchronously create a summary for the server. Contains resource counts for all resources available on the server.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
display |
bool |
whether to display a progress bar |
True |
Returns:
| Type | Description |
|---|---|
ServerSummary |
ServerSummary containing resource counts for all resources available on the server |
__repr__(self)
special
FhirQuerySync (FhirQueryBase)
__init__(self, base_url, resource=None, query_parameters=None, auth=None, headers=None, client=None, output_format='json', proxies=None)
special
Initialize an sync FHIR query object.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
base_url |
str |
Base URL of the FHIR server to query. |
required |
resource |
Union[fhir.resources.fhirresourcemodel.FHIRResourceModel, fhir.resources.core.fhirabstractmodel.FHIRAbstractModel, str] |
Base resource to build the query on. Defaults to None. |
None |
query_parameters |
FhirQueryParameters |
Query parameters object that fully describes a FHIR query. Defaults to None. |
None |
auth |
Auth |
httpx auth object to authenticate the requests. Defaults to None. |
None |
headers |
dict |
Optional additional headers to add to the request. Defaults to None. |
None |
output_format |
str |
Response format of the query. Defaults to "json". |
'json' |
client |
Client |
httpx Client passed from the server. Defaults to None. |
None |
proxies |
Union[str, dict] |
List of proxies to use. Defaults to None. |
None |
all(self, page_callback=None, count=None)
Execute the query and return all results matching the query parameters.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
page_callback |
Union[Callable[[List[fhir.resources.core.fhirabstractmodel.FHIRAbstractModel]], Any], Callable[[], Any]] |
if this argument is set the given callback function will be called for each page of results |
None |
count |
int |
number of results in a page, default value of 50 is used when page_callback is set but no count is |
None |
Returns:
| Type | Description |
|---|---|
QueryResponse |
QueryResponse object containing all resources matching the query, as well os optional included resources. |
limit(self, n, page_callback=None, count=None)
Execute the query and return the first n results matching the query parameters.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
n |
int |
number of resources to return |
required |
page_callback |
Union[Callable[[List[fhir.resources.core.fhirabstractmodel.FHIRAbstractModel]], Any], Callable[[], Any]] |
if this argument is set the given callback function will be called for each page of results |
None |
count |
int |
number of results in a page, default value of 50 is used when page_callback is set but no count is |
None |
Returns:
| Type | Description |
|---|---|
QueryResponse |
QueryResponse object containing the first n resources matching the query, as well os optional included resources. |
first(self)
Return the first resource matching the query parameters.
Returns:
| Type | Description |
|---|---|
QueryResponse |
QueryResponse object containing the first resource matching the query |
count(self)
FhirQueryAsync (FhirQueryBase)
__init__(self, base_url, resource=None, query_parameters=None, auth=None, headers=None, output_format='json', client=None, proxies=None)
special
Initialize an async FHIR query object.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
base_url |
str |
Base URL of the FHIR server to query. |
required |
resource |
Union[fhir.resources.fhirresourcemodel.FHIRResourceModel, fhir.resources.core.fhirabstractmodel.FHIRAbstractModel, str] |
Base resource to build the query on. Defaults to None. |
None |
query_parameters |
FhirQueryParameters |
Query parameters object that fully describes a FHIR query. Defaults to None. |
None |
auth |
Auth |
httpx auth object to authenticate the requests. Defaults to None. |
None |
headers |
dict |
Optional additional headers to add to the request. Defaults to None. |
None |
output_format |
str |
Response format of the query. Defaults to "json". |
'json' |
client |
AsyncClient |
httpx Client passed from the server. Defaults to None. |
None |
proxies |
Union[str, dict] |
List of proxies to use. Defaults to None. |
None |
all(self, page_callback=None, count=None)
async
Execute the query and return all results matching the query parameters.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
page_callback |
Union[Callable[[List[fhir.resources.core.fhirabstractmodel.FHIRAbstractModel]], Any], Callable[[], Any]] |
if this argument is set the given callback function will be called for each page of results |
None |
count |
int |
number of results in a page, default value of 50 is used when page_callback is set but no count is |
None |
Returns:
| Type | Description |
|---|---|
QueryResponse |
QueryResponse object containing all resources matching the query, as well os optional included resources. |
limit(self, n, page_callback=None, count=None)
async
Execute the query and return the first n results matching the query parameters.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
n |
int |
number of resources to return |
required |
page_callback |
Union[Callable[[List[fhir.resources.core.fhirabstractmodel.FHIRAbstractModel]], Any], Callable[[], Any]] |
if this argument is set the given callback function will be called for each page of results |
None |
count |
int |
number of results in a page, default value of 50 is used when page_callback is set but no count is |
None |
Returns:
| Type | Description |
|---|---|
QueryResponse |
QueryResponse object containing the first n resources matching the query, as well os optional included resources. |
first(self)
async
Return the first resource matching the query parameters.
Returns:
| Type | Description |
|---|---|
QueryResponse |
QueryResponse object containing the first resource matching the query |
count(self)
async
Return the number of resources matching the query parameters.
Returns:
| Type | Description |
|---|---|
int |
number of resources matching the query |
QueryResponse
Response object for the results of a FHIR query executed against a FHIR server.
resources: List[fhir.resources.core.fhirabstractmodel.FHIRAbstractModel]
property
readonly
List of primary resources returned by the server.
Returns:
| Type | Description |
|---|---|
List[fhir.resources.core.fhirabstractmodel.FHIRAbstractModel] |
List of FHIRResourceModel objects returned by the server. |
included_resources: List[fhir_kindling.fhir_query.query_response.IncludedResources]
property
readonly
Returns the list of resources included in the search.
Returns:
| Type | Description |
|---|---|
List[fhir_kindling.fhir_query.query_response.IncludedResources] |
List of IncludeResources objects containing the resource type and the list of resources of this type included in the search |
resource_list: List[fhir.resources.core.fhirabstractmodel.FHIRAbstractModel]
property
readonly
Returns a list of all resources returned by the server including the primary resources and the included resources.
Returns:
| Type | Description |
|---|---|
List[fhir.resources.core.fhirabstractmodel.FHIRAbstractModel] |
List of FHIRResourceModel objects returned by the server. |
total: int
property
readonly
Returns the total number of resources returned by the query.
Returns:
| Type | Description |
|---|---|
int |
Total number of resources matching the query. |
__init__(self, response, query_params, output_format=<OutputFormats.JSON: 'json'>, limit=None, count=None)
special
save(self, file_path, output_format='json')
Save the response to a file.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
file_path |
Union[str, pathlib.Path] |
path to the file to save the response to. |
required |
output_format |
str |
output format one of xml|json |
'json' |
Returns:
| Type | Description |
|---|---|
None |
None |