Source code for q2_cores.XP2.core

import logging

from q2_sdk.models.cores.base_core import BaseCore
from q2_sdk.models.cores.models.core_user import CoreUser
from q2_cores.XP2.queries import initial_search_query, demographic_info_query
from q2_cores.XP2.mappers import initial_search_mapper, demographic_info_mapper




class Core(BaseCore):
    CONFIG_FILE_NAME = "XP2_Core"

    OPTIONAL_CONFIGURATIONS = {"GET_ID_BY_SSN": False}

    def __init__(
        self,
        logger: logging.Logger,
        core_user: CoreUser,
        hq_credentials: str = None,
        primary_id: str = None,
        **kwargs,
    ):
        super().__init__(logger, core_user, hq_credentials=hq_credentials, **kwargs)
        self.primary_id = primary_id



    async def set_primary_id(self):
        ind_lookup = getattr(self.config, "GET_ID_BY_SSN", None)
        if ind_lookup:
            initial_mapper = await self.build_initial_ssn_search()
        else:
            initial_mapper = await self.build_initial_search()
        primary_id = await initial_mapper.execute()
        self.primary_id = primary_id




    async def build_initial_search(self) -> initial_search_mapper.InitialSearchMapper:
        initial_search = initial_search_query.InitialSearchQuery(
            self.logger, self.configured_user.customer_primary_cif
        )
        return initial_search_mapper.InitialSearchMapper(
            [initial_search],
            hq_credentials=self.hq_credentials,
            zone_context=self.core_user,
        )




    async def build_initial_ssn_search(
        self,
    ) -> initial_search_mapper.InitialSearchSSNMapper:
        initial_search = initial_search_query.InitialSearchSSNQuery(
            self.logger, self.configured_user.ssn
        )

        return initial_search_mapper.InitialSearchSSNMapper(
            [initial_search],
            hq_credentials=self.hq_credentials,
            zone_context=self.core_user,
        )




    async def build_demographic_info(
        self,
    ) -> demographic_info_mapper.DemographicInfoMapper:
        if not self.primary_id:
            await self.set_primary_id()

        demographic_query = demographic_info_query.DemographicInfoQuery(
            self.logger, self.primary_id
        )

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