import typing
from lxml import objectify
from q2_sdk.models.cores.mappers.demographic_info import BaseDemographicInfoMapper
from q2_sdk.models.demographic import Address, DemographicInfo, Phone, PhoneType
from q2_sdk.models.cores.queries.base_query import BaseQuery
from q2_cores import exceptions
from q2_cores.Users.queries import DemographicInfoQuery
[docs]
class DemographicInfoMapper(BaseDemographicInfoMapper):
[docs]
@staticmethod
def parse_returned_queries(
list_of_queries: typing.List[BaseQuery],
) -> DemographicInfo:
assert len(list_of_queries) == 1, (
"Users only knows how to deal with a single demographicinfo query"
)
assert isinstance(list_of_queries[0], DemographicInfoQuery), (
"Query must be an instance of Users.queries.DemographicInfoQuery"
)
response = list_of_queries[0].raw_core_response
root = objectify.fromstring(response)
# Make check for possible error
status = root.SIGNUPMSGSRSV1.find("USERS.ENROLLTRNRS").STATUS
if status.CODE.pyval != 0:
raise exceptions.CoreException(
"There was a problem with the request to the core. "
"CODE: {}, SEVERITY: {}, USERS.CODE: {}".format(
status.findtext("CODE"),
status.findtext("SEVERITY"),
status.findtext("USERS.CODE"),
)
)
user_node = (
root.SIGNUPMSGSRSV1.find("USERS.ENROLLTRNRS")
.find("USERS.ENROLLRS")
.ENROLLRQ
)
first_name = user_node.FIRSTNAME.text
middle_name = user_node.findtext("MIDDLENAME")
if not middle_name:
first_name_parts = first_name.split(" ")
if len(first_name_parts) > 1:
first_name = first_name_parts[0]
middle_name = first_name_parts[1]
else:
middle_name = ""
last_name = user_node.LASTNAME.text
ssn = "".join(x for x in user_node.TAXID.text if x.isdigit())
dob = user_node.DATEBIRTH.text[:8]
addresses = [
Address(
user_node.ADDR1.text,
"",
user_node.CITY.text,
user_node.STATE.text,
user_node.POSTALCODE.text,
)
]
phones = []
day_phone = user_node.findtext("DAYPHONE")
eve_phone = user_node.findtext("EVEPHONE")
if eve_phone:
phones.append(Phone.build_from_str(eve_phone, PhoneType.PERSONAL))
if day_phone:
phones.append(Phone.build_from_str(day_phone, PhoneType.BUSINESS))
emails = []
email_text = user_node.findtext("EMAIL")
if email_text:
emails.append(email_text)
return DemographicInfo(
dob,
emails,
phones,
addresses,
first_name,
last_name,
ssn,
middle_name=middle_name,
)