from typing import List, Optional
from lxml import objectify
from q2_cores.exceptions import CoreValidationException
from q2_sdk.hq.models.hq_credentials import HqCredentials
from q2_sdk.models.cores.models.core_user import CoreUser
from q2_sdk.models.cores.queries.base_query import BaseQuery
from q2_sdk.models.cores.mappers.base_mapper import BaseMapper
from q2_sdk.models.demographic import DemographicInfo
[docs]
class InitialSearchMapper(BaseMapper):
def __init__(
self,
list_of_queries: List[BaseQuery],
hq_credentials: Optional[HqCredentials] = None,
demo_info: DemographicInfo = None,
zone_context: Optional[CoreUser] = None,
):
super().__init__(
list_of_queries, hq_credentials=hq_credentials, zone_context=zone_context
)
self.demo_info = demo_info
[docs]
def parse_returned_queries(self, list_of_queries: List[BaseQuery]) -> str:
"""
:param list_of_queries: A :class:List of :class:`InitialSearchQuery` to operate against.
:return: a string value of the customer's permanent ID
"""
assert len(list_of_queries) == 1
assert isinstance(list_of_queries[0], BaseQuery), (
"Query must be instance of BaseQuery"
)
root = objectify.fromstring(list_of_queries[0].raw_core_response)
if root.Response.Status.StatusCode.text != "0":
raise CoreValidationException(
f"Core returned unsuccessful status: {root.Response.Status.StatusDesc.text}"
)
if len(root.Response.CustomerSummaryList.CustomerSummary) == 1:
return root.Response.CustomerSummaryList.CustomerSummary.CustomerPermanentId.text
# Multiple demographics were returned, so try to locate the correct one
for summary in root.Response.CustomerSummaryList.CustomerSummary:
customer_permanent_id = summary.CustomerPermanentId.text
if hasattr(summary, "BirthDate"):
core_dob = summary.BirthDate.text
else:
core_dob = ""
normalized_dob = normalize_dob(core_dob)
if (
normalized_dob
and self.demo_info
and normalized_dob == self.demo_info.dob
):
return customer_permanent_id
# Default to return the first user's ID
return (
root.Response.CustomerSummaryList.CustomerSummary.CustomerPermanentId.text
)
[docs]
def normalize_dob(dob: str) -> str:
"""Normalizes the core's DOB string for comparison to the DemographicInfo DOB string
:param dob: DOB string from core
:return: DOB in the format MM-DD-YYYY
"""
if not dob:
return ""
month, day, year = dob.split("/")
if len(month) == 1:
month = f"0{month}"
if len(day) == 1:
day = f"0{day}"
return f"{month}-{day}-{year}"