Source code for fhirpack.pack

import warnings

from typing import Union

import pandas as pd
from pandas import DataFrame

pd.options.mode.chained_assignment = None

from fhirpy import SyncFHIRClient

from fhirpack.auth import AUTH_PARAMS_PRESETS
from fhirpack.auth import Auth
import fhirpack.base
import fhirpack.extraction
import fhirpack.transformation
import fhirpack.load
import fhirpack.custom
import fhirpack.utils as utils
from fhirpack.constants import CONFIG

LOGGER = CONFIG.getLogger(__name__)


class PACK(
    fhirpack.base.BaseMixin,
    fhirpack.extraction.ExtractorMixin,
    fhirpack.transformation.TransformerMixin,
    fhirpack.load.LoaderMixin,
    fhirpack.custom.PluginMixin,
):
    """This class is FHIRPACK's main entry point.
    It is used to connect to the FHIR server, and to interact with the other classes."""

    def __init__(
        self,
        apiBase: str = None,
        client: SyncFHIRClient = None,
        envFile: str = None,
        ignoreEnvFile: bool = False,
        unconnected: bool = False,
        authMethod: str = None,
        authParams: Union[str, dict] = None,
    ):
        """This function initializes the PACK class.

        Args:
            apiBase (Union[str, None]): Base URL of the FHIR server. Defaults to None.
            client (Union[SyncFHIRClient, None]): Client object. Defaults to None.
            envFile (Union[str, None]): Path to the .env file. Defaults to None.
            ignoreEnvFile (Union[bool, None]): If True, the .env file will be ignored. Defaults to False.
            unconnected (Union[bool, None]): If True, the client will not be connected to the server. Defaults to False.
            authMethod (Union[str, None]): Authentication method. Defaults to None.
            authParams (Union[str, dict, None]): Authentication parameters; str with a preset name or dict if individually provided. Defaults to None.
        """
        self.logger = CONFIG.getLogger(__name__)
        self.logger.info("PACK initialization started.")

        if envFile:
            CONFIG.loadConfig(envFile)
        elif not ignoreEnvFile:
            CONFIG.loadConfig()

        if client:
            self.client = client
        elif unconnected:
            self.client = SyncFHIRClient("")
        else:
            self.client = _getConnectedClient(apiBase, authMethod, authParams)

        if self.connected:
            pass
        else:
            self.logger.info("PACK is not connected to server.")
            self.client = SyncFHIRClient("")

        self.logger.info("pack initialization finished")

    def countServerResources(self):
        """This function counts the number of resources on the server.

        Returns:
            DataFrame: DataFrame with columns resourceType and count.
        """

        results = []
        # TODO write function in utils to retrieve current installation path
        # TODO move path with others to common location, CONFIG?
        with open(f"{utils.getInstallationPath()}/data/supported.list") as f:
            while True:
                resource = f.readline().strip()
                if not resource:
                    break

                count = self.client.execute(
                    # TODO handle and test when slash at the end of APIBASE in .env and without
                    # WARNING path must either:
                    # 1. be relative and allow FHIRPy to parametrize the URL
                    # 2. be absolute and contain the params already as they will otherwise be ignored
                    resource,
                    method="get",
                    # if _count has the value 0, this shall be treated the same as _summary=count
                    # https://www.hl7.org/fhir/search.html#count
                    params={"_count": 0},
                ).total
                results.append((resource, count))
        results = DataFrame(results, columns=["resourceType", "count"])
        return results


def _getConnectedClient(apiBase=None, authMethod=None, authParams=None):
    authorization = None

    if apiBase is not None:
        CONFIG.set("APIBASE", apiBase)
        if authMethod:
            CONFIG.set("AUTH_METHOD", authMethod)
        if isinstance(authParams, dict) and authParams:
            pass
        elif isinstance(authParams, str) and authParams:
            authParams = AUTH_PARAMS_PRESETS[authParams]
    else:
        if CONFIG.get("AUTH_PARAMS_PRESET"):
            authParams = AUTH_PARAMS_PRESETS[CONFIG.get("AUTH_PARAMS_PRESET")]
        if CONFIG.get("AUTH_METHOD") == "oauth_password":
            token = Auth.getToken("password", authParams)
            CONFIG.set("OAUTH_TOKEN", token["access_token"]),
            authorization = f"Bearer {token['access_token']}"
        elif CONFIG.get("AUTH_METHOD") == "oauth_token":
            token = CONFIG.get("OAUTH_TOKEN")
            authorization = f"Bearer {token}"
        elif authMethod is None:
            authorization = None
        else:
            raise NotImplementedError

    return SyncFHIRClient(CONFIG.get("APIBASE"), authorization=authorization)