import logging
from typing import Optional, Union
from q2_sdk.core.exceptions import CoreNotConfiguredError
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, Address, Phone
from q2_cores.OSI import mappers, queries, models
[docs]
class UseCode:
"""Enum for determining what AccessNumber means"""
SocialSecurityNumber = 1
AccountNumber = 2
CardNumber = 3
MemberNumber = 4
[docs]
class Core(BaseCore):
CONFIG_FILE_NAME = "OSI_Core"
REQUIRED_CONFIGURATIONS = {"USE_CODE": 1}
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)
access_number = None
valid_use_codes = [1, 2, 3, 4]
use_code = self.config.USE_CODE
if use_code not in valid_use_codes:
self.logger.critical("USE_CODE must be in %s", valid_use_codes)
raise CoreNotConfiguredError
if use_code == UseCode.SocialSecurityNumber:
access_number = core_user.online_user.ssn
self.pin = str(core_user.online_user.ssn)[-4:]
elif use_code == UseCode.MemberNumber:
access_number = core_user.online_user.customer_primary_cif
self.pin = str(core_user.online_user.ssn)[-4:]
elif use_code == UseCode.CardNumber:
access_number = core_user.online_user.customer_primary_cif
self.pin = str(core_user.online_user.customer_primary_cif)[-4:]
elif use_code == UseCode.AccountNumber:
for account in core_user.account_list.accounts:
if (
account.cif_internal_unmasked
== core_user.online_user.customer_primary_cif
and account.access == 7
):
access_number = account.acct_number_internal_unmasked
break
self.pin = str(core_user.online_user.customer_primary_cif)[-4:]
if not access_number and not self.pin:
self.logger.error("Could not set access number or pin for end user")
self.access_number = str(access_number)
self.use_code = use_code
[docs]
async def build_demographic_info(self) -> mappers.DemographicInfoMapper:
query = queries.DemographicInfoQuery(
self.logger, self.access_number, self.pin, self.use_code
)
return mappers.DemographicInfoMapper(
[query], hq_credentials=self.hq_credentials, zone_context=self.core_user
)
[docs]
async def build_update_demographic_info(
self, demographic_info: DemographicInfo
) -> mappers.UpdateDemographicInfoMapper:
update_queries = []
for phone in demographic_info.phones:
try:
phone_type = models.OSIPhoneType(phone.type)
except ValueError:
phone_type = models.OSIPhoneType.Personal
update_queries.append(
queries.UpdatePhoneQuery(
self.logger,
self.access_number,
self.pin,
self.use_code,
phone,
phone_type=phone_type.value,
)
)
for address in demographic_info.addresses:
try:
address_type = models.OSIAddressType(address.address_type)
except ValueError:
address_type = models.OSIAddressType.Primary
update_queries.append(
queries.UpdateAddressQuery(
self.logger,
self.access_number,
self.pin,
self.use_code,
address.city,
address.state,
address.zipcode,
address.address_1,
address.address_2,
address_type=address_type.value,
)
)
return mappers.UpdateDemographicInfoMapper(
update_queries,
hq_credentials=self.hq_credentials,
zone_context=self.core_user,
)
[docs]
async def build_email_info(self) -> mappers.GetEmailsMapper:
query = queries.GetEmailsQuery(
self.logger, self.access_number, self.pin, self.use_code
)
return mappers.GetEmailsMapper(
[query], hq_credentials=self.hq_credentials, zone_context=self.core_user
)
[docs]
async def build_update_email_info(
self, email: str, email_type: models.EmailType
) -> mappers.UpdateEmailInfoMapper:
query = queries.UpdateEmailQuery(
self.logger,
self.access_number,
self.pin,
self.use_code,
email,
email_type.value,
)
return mappers.UpdateEmailInfoMapper(
[query], hq_credentials=self.hq_credentials, zone_context=self.core_user
)
[docs]
async def build_address_update_info(
self, address: Address, address_type: Union[models.OSIAddressType, str]
) -> mappers.UpdateDemographicInfoMapper:
if isinstance(address_type, models.OSIAddressType):
address_type = address_type.value
query = queries.UpdateAddressQuery(
self.logger,
self.access_number,
self.pin,
self.use_code,
address.city,
address.state,
address.zipcode,
address.address_1,
address.address_2,
address_type=address_type,
)
return mappers.UpdateDemographicInfoMapper(
[query], hq_credentials=self.hq_credentials, zone_context=self.core_user
)
[docs]
async def build_phone_update_info(
self, phone: Phone, phone_type: Union[models.OSIPhoneType, str]
) -> mappers.UpdateDemographicInfoMapper:
if isinstance(phone_type, models.OSIPhoneType):
phone_type = phone_type.value
query = queries.UpdatePhoneQuery(
self.logger,
self.access_number,
self.pin,
self.use_code,
phone,
phone_type=phone_type,
)
return mappers.UpdateDemographicInfoMapper(
[query], hq_credentials=self.hq_credentials, zone_context=self.core_user
)