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 q2_cores.Symitar.queries import (
OpenAccountQuery,
GetSubAccountListQuery,
GetSubAccountDetailsQuery,
)
from ...Symitar.models import parser
from ...Symitar.models.open_account_models import SymitarSubAccountDetails
[docs]
class OpenAccountMapper(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], OpenAccountQuery)
response = list_of_queries[0].raw_core_response
symitar_response = parser.parse(response)
new_share_details = objectify.fromstring(symitar_response.payload)
return {"share_id": new_share_details.NEWSHAREID.text}
[docs]
class GetSubAccountDetailsMapper(BaseMapper):
[docs]
@staticmethod
def parse_returned_queries(
list_of_queries: List[BaseQuery],
) -> SymitarSubAccountDetails:
assert len(list_of_queries) == 1
assert isinstance(list_of_queries[0], GetSubAccountDetailsQuery)
response = list_of_queries[0].raw_core_response
symitar_response = parser.parse(response)
share_details = objectify.fromstring(symitar_response.payload)
share_type_id = share_details.SHARETYPE.text
opening_amount = share_details.OPENAMOUNT.text
share_desc = share_details.TYPEDESCRIPTION.text
maturity_date = share_details.MATURITYDATE.text
div_rate = share_details.DIVRATE.text
apy = share_details.APY.text
account_details = SymitarSubAccountDetails(
share_type_id, share_desc, apy, maturity_date, div_rate, opening_amount
)
return account_details
[docs]
class GetSubAccountMapper(BaseMapper):
[docs]
@staticmethod
def parse_returned_queries(list_of_queries: List[BaseQuery]) -> List[dict]:
assert len(list_of_queries) == 1
assert isinstance(list_of_queries[0], GetSubAccountListQuery)
response = list_of_queries[0].raw_core_response
symitar_response = parser.parse(response)
root = objectify.fromstring(symitar_response.payload)
try:
accounts = root.SHARETYPES.SUBTYPE
except AttributeError:
return []
savings = {"name": "Savings", "host_id": "0", "display_order": 0}
checking = {"name": "Checking", "host_id": "1", "display_order": 1}
certificates = {"name": "Certificates", "host_id": "2", "display_order": 2}
for account in accounts:
funding_amount = (
account.MINIMUMBALANCE.text.replace("$", "").replace(",", "").strip()
)
float_funding_amount = float(funding_amount)
if account.SHARECODE.text == "0":
savings.setdefault("available_accounts", []).append({
"desc": account.TYPEDESCRIPTION.text,
"min_funding_amount": "${:,.2f}".format(float_funding_amount),
"float_funding_amount": float_funding_amount,
"account_id": account.TYPENUM.text,
"account_type": account.SHARECODE.text,
})
elif account.SHARECODE.text == "1":
checking.setdefault("available_accounts", []).append({
"desc": account.TYPEDESCRIPTION.text,
"min_funding_amount": "${:,.2f}".format(float_funding_amount),
"float_funding_amount": float_funding_amount,
"account_id": account.TYPENUM.text,
"account_type": account.SHARECODE.text,
})
elif account.SHARECODE.text == "2":
certificates.setdefault("available_accounts", []).append({
"desc": account.TYPEDESCRIPTION.text,
"min_funding_amount": "${:,.2f}".format(float_funding_amount),
"float_funding_amount": float_funding_amount,
"account_id": account.TYPENUM.text,
"account_type": account.SHARECODE.text,
})
accounts = [savings, checking, certificates]
return accounts