import logging
from typing import List
from q2_sdk.models.cores.queries.base_query import BaseQuery
from q2_sdk.models.demographic import DemographicInfo, DriverLicense, Phone, Address
from q2_cores.CorePro.queries import GetCustomerQuery, GetCustomerByEmailQuery
from q2_cores.CorePro.mappers import CoreProBaseMapper
from q2_cores.CorePro.translation_maps import ADDRESS_MAP, PHONE_MAP
BAD_ADDR_MSG = "Address type {} not in known CorePro address types. Skipping"
BAD_PHONE_MSG = "Phone type {} not in known CorePro phone types. Skipping"
ALLOWED_QUERIES = (GetCustomerQuery, GetCustomerByEmailQuery)
[docs]
class GetCustomerMapper(CoreProBaseMapper):
[docs]
async def execute(self) -> DemographicInfo:
return await super().execute()
[docs]
@staticmethod
def get_address_list(raw_addresses: list, logger: logging.Logger) -> List[Address]:
"""
Parses Corepro Address shapes into a list of
`q2_sdk.models.demographic.Address` objects
"""
addr_list = []
for addr in raw_addresses:
if addr["addressType"] not in ADDRESS_MAP:
logger.error(BAD_ADDR_MSG.format(addr["addressType"]))
continue
addr_list.append(
Address(
addr["addressLine1"],
addr["addressLine2"],
addr["city"],
addr["state"],
addr["postalCode"],
ADDRESS_MAP[addr["addressType"]],
)
)
return addr_list
[docs]
@staticmethod
def get_phone_list(raw_phones: list, logger: logging.Logger) -> List[Phone]:
"""
Parses Corepro Phone shapes into a list of
`q2_sdk.models.demographic.Phone` objects
"""
phone_list = []
for phone in raw_phones:
if phone["phoneType"] not in PHONE_MAP:
logger.error(BAD_PHONE_MSG.format(phone["phoneType"]))
continue
phone_list.append(
Phone.build_from_str(phone["number"], PHONE_MAP[phone["phoneType"]])
)
return phone_list
[docs]
@staticmethod
def parse_returned_queries(list_of_queries: List[BaseQuery]) -> DemographicInfo:
assert any(isinstance(list_of_queries[0], x) for x in ALLOWED_QUERIES), (
"Query must be an instance of CorePro.queries.GetCustomerQuery"
)
response_obj = CoreProBaseMapper.parse_returned_queries(list_of_queries)
logger = list_of_queries[0].logger
data = response_obj.data
address_list = GetCustomerMapper.get_address_list(data["addresses"], logger)
phone_list = GetCustomerMapper.get_phone_list(data["phones"], logger)
response = DemographicInfo(
data["birthDate"],
[data["emailAddress"]],
phone_list,
address_list,
data["firstName"],
data["lastName"],
data.get("taxId", data.get("taxIdMasked")),
middle_name=data["middleName"],
driver_license=DriverLicense(
data.get(
"driversLicenseNumber", data.get("driversLicenseNumberMasked")
),
data["driversLicenseState"],
),
primary_cif=data["customerId"],
)
return response