Source code for q2_cores.FiservCleartouch.mappers.initial_search_mapper

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}"