Source code for q2_cores.EPL.core

import logging
from typing import Optional

from q2_sdk.hq.models.hq_credentials import HqCredentials
from q2_sdk.models.cores.base_core import BaseCore
from q2_sdk.models.cores.models.core_user import CoreUser
from q2_cores.EPL.queries.demographic_info_query import DemographicInfoQuery
from q2_cores.EPL.mappers.demographic_info_mapper import DemographicInfoMapper




class Core(BaseCore):
    CONFIG_FILE_NAME = "EPL_Core"
    REQUIRED_CONFIGURATIONS = {"INSTITUTION_ID": "01", "CONSUMER_ID": "Q2"}
    OPTIONAL_CONFIGURATIONS = {"REQUEST_NAMESPACE": "", "HEADER_NAMESPACE": ""}

    def __init__(
        self,
        logger: logging.Logger,
        core_user: CoreUser,
        hq_credentials: Optional[HqCredentials] = None,
        **kwargs,
    ):
        super().__init__(logger, core_user, hq_credentials=hq_credentials, **kwargs)



    async def build_demographic_info(self) -> DemographicInfoMapper:
        request_namespace = getattr(self.config, "REQUEST_NAMESPACE", None)
        header_namespace = getattr(self.config, "HEADER_NAMESPACE", None)
        demographic_query = DemographicInfoQuery(
            self.logger,
            self.configured_user.ssn,
            self.config.INSTITUTION_ID,
            self.config.CONSUMER_ID,
            request_namespace=request_namespace,
            header_namespace=header_namespace,
        )

        return DemographicInfoMapper(
            [demographic_query],
            hq_credentials=self.hq_credentials,
            zone_context=self.core_user,
        )