"""Bindings for Q2's core. Docs: https://docs.corepro.io/api/overview"""
import logging
from typing import Optional, Union
from q2_sdk.core.configuration import EnvVar
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_sdk.models.demographic import DemographicInfo
from q2_cores.CorePro import mappers, queries
[docs]
class Core(BaseCore):
CONFIG_FILE_NAME = "CorePro_Core"
REQUIRED_CONFIGURATIONS = {
"USE_BRIDGE": False,
"API_KEY": EnvVar("Q2SDK_COREPRO_API_KEY"),
"API_SECRET": EnvVar("Q2SDK_COREPRO_API_SECRET"),
"BASE_URL_OVERRIDE": EnvVar("Q2SDK_COREPRO_OVERRIDE_URL"),
}
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)
[docs]
async def build_base_query(
self,
url_path: str,
verb: str,
body: Optional[dict] = None,
mock_response: Optional[Union[str, dict]] = None,
) -> mappers.CoreProBaseMapper:
"""
Generic call builder for CorePro as per the docs: https://docs.corepro.io/api/overview
:param url_path: URL according to CorePro docs (/customer/get/id for instance)
:verb: GET, POST, PUT, etc.
:body: (Optional) JSON payload
"""
query = queries.base_query.CoreProBaseQuery(
self.logger, url_path, verb, body=body, mock_response=mock_response
)
return mappers.CoreProBaseMapper(
[query],
self.config,
hq_credentials=self.hq_credentials,
zone_context=self.core_user,
)
[docs]
async def build_get_customer(self) -> mappers.GetCustomerMapper:
query = queries.GetCustomerQuery(
self.logger, self.configured_user.customer_primary_cif
)
return mappers.GetCustomerMapper(
[query],
self.config,
hq_credentials=self.hq_credentials,
zone_context=self.core_user,
)
[docs]
async def build_demographic_info(self) -> mappers.GetCustomerMapper:
"""Alias for CorePro's get_customer"""
return await self.build_get_customer()
[docs]
async def build_get_customer_by_email(self) -> mappers.GetCustomerMapper:
query = queries.GetCustomerByEmailQuery(
self.logger, self.configured_user.email_address
)
return mappers.GetCustomerMapper(
[query],
self.config,
hq_credentials=self.hq_credentials,
zone_context=self.core_user,
)
[docs]
async def build_update_demographic_info(
self, demographic_info: DemographicInfo
) -> mappers.CoreProBaseMapper:
query = queries.UpdateDemographicInfoQuery(
self.logger, self.configured_user.customer_primary_cif, demographic_info
)
return mappers.CoreProBaseMapper(
[query],
self.config,
hq_credentials=self.hq_credentials,
zone_context=self.core_user,
)