from typing import List, Optional
from lxml import objectify
from q2_sdk.core import cache
from q2_sdk.hq.models.hq_credentials import HqCredentials
from q2_sdk.models.cores.mappers.base_mapper import BaseMapper
from q2_sdk.models.cores.queries.base_query import BaseQuery
from q2_sdk.models.cores.models.core_user import CoreUser
from ..queries import GetSubAccountDetailsQuery, GetSubAccountListQuery
from ...Users.models.open_account_models import UsersSubAccountDetails
[docs]
class GetSubAccountMapper(BaseMapper):
def __init__(
self,
list_of_queries: List[BaseQuery],
cache_key: str,
hq_credentials: Optional[HqCredentials] = None,
zone_context: Optional[CoreUser] = None,
):
super().__init__(
list_of_queries, hq_credentials=hq_credentials, zone_context=zone_context
)
self.cache_key = cache_key
[docs]
async def execute(self) -> List[dict]:
parsed_queries = await super().execute()
if self.cache_key:
get_sub_account_query = self.list_of_queries[0]
try:
cache.get_cache().set(
self.cache_key, get_sub_account_query.raw_core_response, expire=120
)
except Exception:
get_sub_account_query.logger.warning(
"Unable to connect to cache. Will need to requery core"
)
return parsed_queries
[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
accounts_response = objectify.fromstring(response)
account_groups = accounts_response.findall(".//ACNGROUP")
accounts = {}
for account_group in account_groups:
group_name = account_group.GROUPNAME.text
description = account_group.SUBGROUP.SUBGROUPNAME.text
account = account_group.SUBGROUP.OPENACN
account_type = account.ACNTYPE.text
type_element = account.find("TYPE")
if type_element:
host_account_type = account.TYPE.text
else:
host_account_type = account.CLA.text
float_funding_amount = float(account.MIN.text)
accounts.setdefault(group_name, []).append({
"desc": description,
"min_funding_amount": "${:,.2f}".format(float_funding_amount),
"float_funding_amount": float_funding_amount,
"account_id": host_account_type,
"account_type": account_type,
})
arranged_accounts = []
for i, account in enumerate(accounts):
host_id = accounts[account][0]["account_type"]
arranged_accounts.append({
"name": account,
"host_id": host_id,
"display_order": i,
"available_accounts": accounts[account],
})
return arranged_accounts
[docs]
class GetSubAccountDetailsMapper(BaseMapper):
[docs]
async def execute(self) -> UsersSubAccountDetails:
"""
Overwriting execute to only return the parsed response of the sub_account_response from cache
No need to make a core call here
:return:
"""
details_query = self.list_of_queries[0]
if not details_query.raw_core_response:
return await super().execute()
return self.parse_returned_queries(self.list_of_queries)
[docs]
@staticmethod
def parse_returned_queries(
list_of_queries: List[GetSubAccountDetailsQuery],
) -> UsersSubAccountDetails:
sub_account_details_query = list_of_queries[0]
assert isinstance(sub_account_details_query, GetSubAccountDetailsQuery)
selected_account_id = sub_account_details_query.account_id
selected_account_type = sub_account_details_query.account_type
funding_amount = str(sub_account_details_query.funding_amount)
sub_account_response = sub_account_details_query.raw_core_response
accounts_response = objectify.fromstring(sub_account_response)
account_groups = accounts_response.findall(".//ACNGROUP")
for account_group in account_groups:
account = account_group.SUBGROUP.OPENACN
account_type = account.ACNTYPE.text
type_element = account.find("TYPE")
if type_element:
host_account_type = account.TYPE.text
else:
host_account_type = account.CLA.text
if (
selected_account_type == account_type
and selected_account_id == host_account_type
):
description = account_group.SUBGROUP.SUBGROUPNAME.text
apy = account.APYRATE.text
maturity_date = account.find("MATDATE")
if maturity_date:
maturity_date = account.MATDATE.text
else:
maturity_date = None
div_rate = account.find("RATE")
if div_rate:
div_rate = account.RATE.text
account_details = UsersSubAccountDetails(
host_account_type,
description,
apy,
maturity_date,
div_rate,
funding_amount,
)
return account_details