from argparse import _SubParsersAction
from functools import partial
from typing import Optional
from lxml import objectify
from q2_sdk.core.dynamic_imports import (
api_ExecuteStoredProcedure as ExecuteStoredProcedure,
)
from q2_sdk.hq import api_helpers
from .db_object import DbObject
[docs]
class LinkedAccount(DbObject):
"""
Can link and unlink existing accounts in the Q2 product to customers and users. Combines the Q2_CustomerAccount and
Q2_UserAccount tables
"""
[docs]
def add_arguments(self, parser: _SubParsersAction):
subparser = parser.add_parser("get_linked_accounts")
subparser.set_defaults(parser="get_linked_accounts")
subparser.set_defaults(func=partial(self.get, serialize_for_cli=True))
subparser.add_argument("user_id", help="Q2_UserAccount.UserID")
subparser = parser.add_parser("get_customer_linked_accounts")
subparser.set_defaults(parser="get_customer_linked_accounts")
subparser.set_defaults(
func=partial(self.get_customer_accounts, serialize_for_cli=True)
)
subparser.add_argument("customer_id", help="Q2_Customer.CustomerID")
subparser = parser.add_parser("remove_linked_account")
subparser.set_defaults(parser="remove_linked_account")
subparser.set_defaults(func=partial(self.delete))
subparser.add_argument("user_id", help="Q2_UserAccount.UserID")
subparser.add_argument("host_account_id", help="Q2_UserAccount.HostAccountID")
subparser = parser.add_parser("add_linked_account")
subparser.set_defaults(parser="add_linked_account")
subparser.set_defaults(func=partial(self.add))
subparser.add_argument("customer_id", help="Q2_Customer.CustomerID")
subparser.add_argument("user_id", help="Q2_User.UserID")
subparser.add_argument(
"account_access",
help="1=Deposit 2=View 4=Withdraw (combine for added rights)",
)
subparser.add_argument("host_account_id", help="Q2_HostAccount.HostAccountID")
[docs]
async def get(self, user_id: int, serialize_for_cli=False):
response = await self.call_hq(
"sdk_GetLinkedAccounts",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int, "UserID", user_id
)
]),
)
if serialize_for_cli:
response = self.serialize_for_cli(
response,
[
"UserAccountID",
"UserID",
"CustomerAccountID",
"Access",
"HostAccountID",
],
)
return response
[docs]
async def get_customer_accounts(self, customer_id: int, serialize_for_cli=False):
response = await self.call_hq(
"sdk_GetCustomerLinkedAccount",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int, "CustomerID", customer_id
)
]),
)
if serialize_for_cli:
response = self.serialize_for_cli(
response,
[
"CustomerAccountID",
"CustomerID",
"HostInternalCIF",
"HostAccountID",
"Access",
"PrimaryOnly",
],
)
return response
[docs]
async def add(
self,
customer_id: int,
user_id: int,
account_access: int,
host_account_id: int,
user_role_id: Optional[int] = None,
):
customer_id = int(customer_id) if isinstance(customer_id, str) else customer_id
user_id = int(user_id) if isinstance(user_id, str) else user_id
account_access = (
int(account_access) if isinstance(account_access, str) else account_access
)
host_account_id = (
int(host_account_id)
if isinstance(host_account_id, str)
else host_account_id
)
request_xml = await self._build_account_link_xml(
customer_id,
user_id,
account_access,
host_account_id=host_account_id,
user_role_id=user_role_id,
)
await self.call_hq(
"apispAddAccountAssociation",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Xml,
"accountAssociationData",
request_xml,
)
]),
)
inner_result = objectify.fromstring(
self.hq_response.result_node.find(".//Table/ResultInfo").text
)
error = self.hq_response.get_hq_return_error(inner_result)
if error:
return "ERROR: {}".format(error)
[docs]
async def add_by_primary_cif(
self,
customer_id: int,
user_id: int,
account_access: int,
primary_cif: str,
primary_only=True,
):
request_xml = await self._build_account_link_xml(
customer_id,
user_id,
account_access,
primary_cif=primary_cif,
primary_only=primary_only,
)
await self.call_hq(
"apispAddAccountAssociation",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Xml,
"accountAssociationData",
request_xml,
)
]),
)
inner_result = objectify.fromstring(
self.hq_response.result_node.find(".//Table/ResultInfo").text
)
error = self.hq_response.get_hq_return_error(inner_result)
if error:
return "ERROR: {}".format(error)
async def _build_account_link_xml(
self,
customer_id,
user_id,
account_access,
primary_cif=None,
host_account_id=None,
primary_only=True,
user_role_id: Optional[int] = None,
):
customer_accounts = await self.get_customer_accounts(customer_id)
access = None
customer_account_id = None
for row in customer_accounts:
if host_account_id and row.findtext("HostAccountID") == str(
host_account_id
):
customer_account_id = int(row.CustomerAccountID.text)
access = int(row.Access.text)
break
elif primary_cif and row.findtext("HostInternalCIF") == primary_cif:
customer_account_id = int(row.CustomerAccountID.text)
access = int(row.Access.text)
break
if access and account_access > access:
account_access = access
self.logger.info(
"Given access %s is higher than customer access %s. Not allowed.",
account_access,
access,
)
if host_account_id:
add_account_request = api_helpers.build_account_association_xml(
customer_id,
user_id,
account_access,
host_account_id=host_account_id,
customer_account_id=customer_account_id,
primary_only=primary_only,
user_role_id=user_role_id,
)
else:
add_account_request = api_helpers.build_account_association_xml(
customer_id,
user_id,
account_access,
cif_internal=primary_cif,
link_by_cif=True,
customer_account_id=customer_account_id,
primary_only=primary_only,
)
return add_account_request
[docs]
async def delete(self, user_id: int, host_account_id: int):
await self.call_hq(
"sdk_UnlinkAccount",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int, "userID", user_id
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int,
"HostAccountID",
host_account_id,
),
]),
)