Source code for q2_sdk.models.cores.base_core

from typing import Optional

from q2_sdk.core import configuration
from q2_sdk.hq.models.hq_credentials import HqCredentials
from q2_sdk.hq.models.online_user import OnlineUser

from .mappers.demographic_info import BaseDemographicInfoMapper
from .mappers.card_list import BaseCardListMapper
from .mappers.update_demographic_info import BaseUpdateDemographicInfoMapper
from .models.core_user import CoreUser
from ..demographic import DemographicInfo
from ...core.q2_logging.logger import Q2LoggerType




class BaseCore:
    """
    If REQUIRED_CONFIGURATIONS is set, ensures the entries are set in the
    core's settings file or the web server will not start.

    REQUIRED_CONFIGURATIONS is a dictionary of key value pairs where the
    key is the required name and the value is the default value when
    ``q2 generate_config`` is called.

    OPTIONAL_CONFIGURATIONS is identical to REQUIRED_CONFIGURATIONS, but
    will not prevent the web server from starting if missing.
    """

    CONFIG_FILE_NAME = ""
    REQUIRED_CONFIGURATIONS = {}
    OPTIONAL_CONFIGURATIONS = {}

    def __init__(
        self,
        logger: Q2LoggerType,
        core_user: CoreUser,
        hq_credentials: Optional[HqCredentials] = None,
        db_config_dict: dict | None = None,
    ):
        """
        All cores will inherit from this common base class. This contains a
        self.configured_user property, which is an instance of
        `q2_sdk.models.cores.models.core_user.CoreUser`.
        """

        if db_config_dict:
            self.config = configuration.Configuration(
                None,
                self.REQUIRED_CONFIGURATIONS,
                self.OPTIONAL_CONFIGURATIONS,
                db_config_dict,
            )
        else:
            self.config = configuration.get_configuration(
                self.CONFIG_FILE_NAME, self.REQUIRED_CONFIGURATIONS
            )
        self.logger = logger
        self.core_user = core_user
        if isinstance(core_user, CoreUser):
            self.configured_user = core_user.online_user
        elif isinstance(core_user, OnlineUser):
            self.configured_user = core_user
        else:
            raise TypeError(
                "core_user argument must be either an instance of CoreUser or OnlineUser"
            )
        self.hq_credentials = hq_credentials



    async def build_demographic_info(self) -> BaseDemographicInfoMapper:
        """Returns a DemographicInfoMapper ready to execute"""
        raise NotImplementedError




    async def build_update_demographic_info(
        self, demographic_info: DemographicInfo
    ) -> BaseUpdateDemographicInfoMapper:
        """Returns a UpdateDemographicInfoMapper ready to execute"""
        raise NotImplementedError




    async def build_get_cards(self) -> BaseCardListMapper:
        """Returns a CardListMapper ready to execute"""
        raise NotImplementedError




    def update_configured_user(self, user: OnlineUser):
        """
        If it is necessary to call the core for a user OTHER than the current online_user,
        first call this method with a new q2_sdk.hq.models.OnlineUser instance
        """

        assert isinstance(user, OnlineUser)
        self.configured_user = user


    def get_payment_builder(self, payment_type: str):
        raise NotImplementedError(
            "{} is not a payment supported by this core".format(payment_type)
        )