FHIR

THis module contains the FHIR related functionality of the PHT, which includes generating query URLs, executing the query against a FHIR server. It will also store the results in a selected format and check parsed results for k-anonymity.

FHIR Client

Client for interacting with FHIR servers following the v4 FHIR specifications.

class train_lib.fhir.fhir_client.BearerAuth(token)[source]: Bases: requests.auth.AuthBase

class train_lib.fhir.fhir_client.PHTFhirClient(server_url: Optional[str] = None, username: Optional[str] = None, password: Optional[str] = None, token: Optional[str] = None, fhir_server_type: Optional[str] = None, disable_auth: bool = False, disable_k_anon: bool = False)[source]

Bases: object

async execute_query(query_file: Optional[Union[str, os.PathLike, _io.BytesIO]] = None, query: Optional[dict] = None, store_results: bool = False) → pandas.core.frame.DataFrame[source]: Asynchronously build the query string and execute it against the given fhir server either based on a query.json file or based on a dictionary containing the query file content. :param query_file: definition of the query given in json format either a string, a path to a file or in memory file object :param query: dictionary containing query definition :return:

health_check()[source]

static read_query_file(file: Union[str, os.PathLike, _io.BytesIO]) → dict[source]: Reads the content of a query.json file either given as a path or as in memory file object :param file: path to file or in memory file object to parse :return: dictionary representation of the query.json file

store_query_results(query_results, filename: str, storage_dir: Optional[str] = None) → str[source]

Store the results from the query according to the output format specified in the query file

Parameters

query_results – The parsed or raw response from a fhir server to the given query
filename – name for the results file (defined in query.json)
storage_dir – directory in which the data should be stored

Returns

upload_resource_or_bundle(resource=None, bundle: Optional[fhir.resources.bundle.Bundle] = None)[source]: Upload a fhir resources bundle or single resource in json format to the FHIR client associated with this instance :param resource: FHIR resource in json format :param bundle: FHIR Bundle containing multiple resources grouped in a transaction :return:

Query building

train_lib.fhir.fhir_query_builder.build_query_string(query_dict: dict) → str[source]

Builds a valid query string to perform a get request against a fhir server based on the given dictionary containing the definition of a fhir query in json format.

Parameters: query_dict – dictionary defining fhir search parameters
Returns

train_lib.fhir.fhir_query_builder.load_query_file(query_json: Union[str, os.PathLike, bytes]) → dict[source]: Load a give json file defining the fhir query into a dictionary :param query_json: :return:

train_lib.fhir.fhir_query_builder.process_main_resource_parameters(resource_params: List[dict]) → str[source]: Build the query parameters to be applied directly to queried fhir resource. :param resource_params: List of dictionary containing the name of the parameter to search and the search condition :return: part of

train_lib.fhir.fhir_query_builder.process_reverse_chain_params(resource: str, reverse_chains: List[dict]) → str[source]: Creates a query string based on the given reverse chain parameters for the queried resource (querying based on other resources that refer to the resource. :param resource: the main resource on which to query references :param reverse_chains: list of reverse chain resources and parameters :return: query string to use in fhir search

k-anonymity

train_lib.fhir.fhir_k_anonymity.anonymize(df: pandas.core.frame.DataFrame, k: int = 3, id_cols: Optional[List[str]] = None) → pandas.core.frame.DataFrame[source]

Attempts to generalize the given dataframe to make it k-anonymized

Parameters

df – dataframe to check
k –
id_cols – optional parameter specifying a subset of columns in the dataframe to generalize

Returns

train_lib.fhir.fhir_k_anonymity.generalize_datetime_column(date_col: pandas.core.series.Series, level: int = 2)[source]

train_lib.fhir.fhir_k_anonymity.generalize_numeric_column(num_col: pandas.core.series.Series)[source]

train_lib.fhir.fhir_k_anonymity.is_k_anonymized(df: pandas.core.frame.DataFrame, k: int = 3, id_cols: Optional[List[str]] = None)[source]: Checks if a dataframe satisfies k-anonymity for the given k. If id_cols is given only these columns are checked :param df: dataframe to check for k-anonymity :param k: the number samples that need to have the same values :param id_cols: optional subset of columns in the dataframe that are exclusively checked for k-anonymity :return: boolean indicating wether the dataframe satisfies k-anonymity