from typing import List
from lxml import objectify
from q2_sdk.models.cores.mappers.base_mapper import BaseMapper
from q2_sdk.models.cores.queries.base_query import BaseQuery
from ...Symitar.queries import SymitarBaseQuery
from ...Symitar.models import parser
SYMITAR_MAPPINGS = {
"cif": "CIF",
"loan_id": "ID",
"type": "TYPE",
"dob": "BIRTHDATE",
"description": "DESCRIPTION",
"due_date": "DUEDATE",
"payment_amount": "PAYMENT",
"balance": "BALANCE",
}
[docs]
class GetEligibleSkipLoansMapper(BaseMapper):
[docs]
@staticmethod
def parse_returned_queries(list_of_queries: List[BaseQuery]) -> dict:
assert len(list_of_queries) == 1
assert isinstance(list_of_queries[0], SymitarBaseQuery), (
"Query must be an instance of Symitar.queries.SymitarBaseQuery"
)
response = list_of_queries[0].raw_core_response
symitar_response = parser.parse(response)
root = objectify.fromstring(symitar_response.payload)
eligible_loans = root.findall(".//LOANS/LOAN")
ineligible_loans = root.findall(".//LOANS/NOTELIGIBLELOAN")
eligible_loan_models = []
ineligible_loan_models = []
for loan in eligible_loans:
loan_data = [
(x, str(getattr(loan, SYMITAR_MAPPINGS[x])))
for x in SYMITAR_MAPPINGS
if hasattr(loan, SYMITAR_MAPPINGS[x])
]
loan_model = dict(loan_data)
eligible_loan_models.append(loan_model)
for loan in ineligible_loans:
loan_data = [
(x, str(getattr(loan, SYMITAR_MAPPINGS[x])))
for x in SYMITAR_MAPPINGS
if hasattr(loan, SYMITAR_MAPPINGS[x])
]
loan_model = dict(loan_data)
if hasattr(loan, "FAILEDREASONS"):
ineligible_reasons = {}
reasons = [
(reason.tag, reason.text)
for reason in getattr(loan, "FAILEDREASONS").iterchildren()
]
ineligible_reasons["ineligible_reasons"] = dict(reasons)
loan_model.update(ineligible_reasons)
ineligible_loan_models.append(loan_model)
return {
"eligible_loans": eligible_loan_models,
"ineligible_loans": ineligible_loan_models,
}