import logging
from lxml import etree
from q2_sdk.models.cores.queries.base_query import BaseQuery
from ... import data_helpers
from ..queries.mock_responses import (
demographic_info_response,
mock_person_verification_response,
)
from ..utils import SearchType
[docs]
class DemographicInfoQuery(BaseQuery):
"""
Builds the payload for the demographic call to the Corelation Core. Options to search include: login verification or person
verification method
"""
def __init__(
self,
logger: logging.Logger,
person_serial=None,
req_dict: dict = None,
search_type: SearchType = SearchType.LOGIN_VERIFY,
):
self.logger = logger
self.req_dict = req_dict
self.search_type = search_type
if not self.req_dict and self.search_type == SearchType.LOGIN_VERIFY:
self.req_dict = {"person_serial": person_serial}
super().__init__(logger)
[docs]
def build(self):
"""
Creates query to get demographic info.
Example query to search using login verification:
.. code-block:: xml
<query xmlns="http://www.corelationinc.com/queryLanguage/v1.0">
<sequence>
<transaction>
<step>
<search>
<tableName>PERSON</tableName>
<filterName>BY_SERIAL</filterName>
<formatName>VERIFICATION</formatName>
<includeTotalHitCount option="Y"/>
<returnLimit>10</returnLimit>
<parameter>
<columnName>SERIAL</columnName>
<contents>36323</contents>
</parameter>
</search>
</step>
</transaction>
</sequence>
</query>
Example query to search using person verification:
This person verification call with TIN will help to get the personal serial associated with the TIN even if
login record is not setup in keystone
.. code-block:: xml
<query xmlns="http://www.corelationinc.com/queryLanguage/v1.0">
<sequence>
<transaction>
<step>
<search>
<tableName>PERSON</tableName>
<filterName>BY_TIN</filterName>
<parameter>
<columnName>TIN</columnName>
<contents>123456789</contents>
</parameter>
<formatName>VERIFICATION</formatName>
<includeTotalHitCount option="Y"/>
<returnLimit>50</returnLimit>
<includeSelectColumns option="N"/>
</search>
</step>
</transaction>
</sequence>
</query>
"""
query_node = etree.Element(
"query", xmlns="http://www.corelationinc.com/queryLanguage/v1.0"
)
sequence_node = etree.SubElement(query_node, "sequence")
transaction_node = etree.SubElement(sequence_node, "transaction")
step_node = etree.SubElement(transaction_node, "step")
search_node = etree.SubElement(step_node, "search")
etree.SubElement(search_node, "tableName").text = "PERSON"
if self.search_type == SearchType.LOGIN_VERIFY:
etree.SubElement(search_node, "filterName").text = "BY_SERIAL"
etree.SubElement(search_node, "formatName").text = "VERIFICATION"
etree.SubElement(search_node, "includeTotalHitCount", option="Y")
etree.SubElement(search_node, "returnLimit").text = "10"
parameter_node = etree.SubElement(search_node, "parameter")
etree.SubElement(parameter_node, "columnName").text = "SERIAL"
etree.SubElement(parameter_node, "contents").text = str(
self.req_dict["person_serial"]
)
else: # search type is SearchType.PERSON_VERIFICATION
etree.SubElement(search_node, "filterName").text = self.req_dict[
"verify_by"
]
parameter_node = etree.SubElement(search_node, "parameter")
if self.req_dict["verify_by"] == "BY_TIN":
etree.SubElement(parameter_node, "columnName").text = "TIN"
etree.SubElement(parameter_node, "contents").text = self.req_dict[
"login_id"
]
etree.SubElement(search_node, "formatName").text = "VERIFICATION"
etree.SubElement(search_node, "includeTotalHitCount", option="Y")
etree.SubElement(search_node, "returnLimit").text = "50"
etree.SubElement(search_node, "includeSelectColumns", option="N")
xml_payload = etree.tostring(query_node, pretty_print=False).decode()
return data_helpers.normalize_xml_str(xml_payload)
[docs]
def mock_response(self):
if self.search_type == SearchType.LOGIN_VERIFY:
return demographic_info_response()
return mock_person_verification_response()