from argparse import _SubParsersAction
from enum import Enum
from functools import partial
from dataclasses import dataclass
from typing import Optional
from lxml.objectify import IntElement, StringElement, BoolElement, E, fromstring
from lxml.etree import tostring
from q2_sdk.core.dynamic_imports import (
api_ExecuteStoredProcedure as ExecuteStoredProcedure,
)
from q2_sdk.core.exceptions import DatabaseDataError
from q2_sdk.hq.db.audit_record import AuditRecord
from q2_sdk.hq.models.hq_params.stored_procedure import Param
from q2_sdk.tools.decorators import dev_only
from .db_object import DbObject
from .representation_row_base import RepresentationRowBase
D_TYPES = ExecuteStoredProcedure.DataType
[docs]
class HostAccountRow(RepresentationRowBase):
HostAccountID: IntElement = "HostAccountID"
AccountNumberInternal: StringElement = "AccountNumberInternal"
AccountNumberExternal: StringElement = "AccountNumberExternal"
ProductID: IntElement = "ProductID"
ProductName: StringElement = "ProductName"
ProductNickName: StringElement = "ProductNickName"
AccountDesc: StringElement = "AccountDesc"
CIFInternal: StringElement = "CIFInternal"
CIFExternal: StringElement = "CIFExternal"
DefaultPwd: StringElement = "DefaultPwd"
DataAsOfDate: StringElement = "DataAsOfDate"
BranchID: IntElement = "BranchID"
SkipBalanceCheck: BoolElement = "SkipBalanceCheck"
ABA: StringElement = "ABA"
IsExternalAccount: BoolElement = "IsExternalAccount"
AccountInfo: StringElement = "AccountInfo"
LastStatusChange: StringElement = "LastStatusChange"
AccountStatusID: IntElement = "AccountStatusID"
LastHistoryDate: StringElement = "LastHistoryDate"
[docs]
class HostAccountCustomerRow(RepresentationRowBase):
CustomerID: IntElement = "CustomerID"
UserID: IntElement = "UserID"
HostAccountID: IntElement = "HostAccountID"
AccountNumberInternal: StringElement = "AccountNumberInternal"
AccountNumberExternal: StringElement = "AccountNumberExternal"
ProductID: IntElement = "ProductID"
ProductName: StringElement = "ProductName"
AccountDesc: StringElement = "AccountDesc"
CIFInternal: StringElement = "CIFInternal"
CIFExternal: StringElement = "CIFExternal"
DataAsOfDate: StringElement = "DataAsOfDate"
ProductTypeName: StringElement = "ProductTypeName"
ABA: StringElement = "ABA"
IsExternalAccount: BoolElement = "IsExternalAccount"
[docs]
class HostAccountCustomerWithProductRow(RepresentationRowBase):
CustomerID: IntElement = "CustomerID"
HostAccountID: IntElement = "HostAccountID"
ProductName: StringElement = "ProductName"
ProductID: IntElement = "ProductID"
AccountStatus: StringElement = "AccountStatus"
[docs]
@dataclass
class UpdateAccountRightsParameters:
hostAccountId: int
userId: int
can_view: bool
can_deposit: bool
can_withdraw: bool
[docs]
def calculate_access(self) -> int:
access = 0
access += 1 if self.can_deposit else 0
access += 2 if self.can_view else 0
access += 4 if self.can_withdraw else 0
return access
[docs]
def build_sql_parameters(self):
params = []
for param in [
Param(self.hostAccountId, D_TYPES.Int, "hostAccountId"),
Param(self.userId, D_TYPES.Int, "userId"),
Param(self.calculate_access(), D_TYPES.Int, "access"),
]:
param.add_to_param_list(params)
return params
[docs]
@dataclass
class ProductChangeParameters:
host_account_id: int
old_product_id: int
new_product_id: int
[docs]
def build_xml_node(self):
return E.AccountRecord(
HostAccountID=str(self.host_account_id),
OldProductID=str(self.old_product_id),
NewProductID=str(self.new_product_id),
)
[docs]
class ProductChangeStatus(Enum):
Ok = "OK"
AllRecordsSkipped = "AllRecordsSkipped"
PartialUpdate = "PartialUpdate"
[docs]
@dataclass
class ProductChangeResults:
result: ProductChangeStatus
successful_accounts: list
skipped_accounts: list
[docs]
class HostAccount(DbObject):
REPRESENTATION_ROW_CLASS = HostAccountRow
[docs]
def add_arguments(self, parser: _SubParsersAction):
subparser = parser.add_parser("get_host_account_by_id")
subparser.set_defaults(parser="get_host_account_by_id")
subparser.set_defaults(func=partial(self.get_by_id, serialize_for_cli=True))
subparser.add_argument("host_account_id", help="Q2_HostAccount.HostAccountID")
[docs]
@dev_only
async def check_adapter_mode(self):
"""Note: this only works in the dev environment"""
return await self.call_hq(
"sdk_SetToBatch", ExecuteStoredProcedure.SqlParameters([])
)
[docs]
@dev_only
async def update_to_batch(self):
"""Note: this only works in the dev environment"""
return await self.call_hq(
"sdk_SetToBatch",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Bit, "update_to_batch", True
)
]),
)
[docs]
async def get_by_id(
self, host_account_id: int, serialize_for_cli=False
) -> HostAccountRow:
response = await self.call_hq(
"sdk_GetHostAccountById",
sql_parameters=ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int,
"host_account_id",
host_account_id,
)
]),
)
if response:
response = response[0]
if serialize_for_cli:
response = self.serialize_for_cli(
[response],
fields_to_display=[
"AccountNumberInternal",
"AccountNumberExternal",
"ProductID",
"CIFInternal",
"CIFExternal",
"DataAsOfDate",
"IsExternalAccount",
"ProductName",
],
)
return response
[docs]
@dev_only
async def create(
self,
account_internal,
account_external,
product_id,
cif,
account_description,
is_external: bool = False,
) -> list[HostAccountRow]:
"""Note: this only works in the dev environment"""
return await self.call_hq(
"sdk_AddHostAccount",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"acct_internal",
account_internal,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"acct_external",
account_external,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int, "product_id", product_id
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar, "cif", cif
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"acct_description",
account_description,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Bit, "is_external", is_external
),
]),
)
[docs]
async def search_by_external_number(
self, external_account_number
) -> list[HostAccountRow]:
return await self.call_hq(
"sdk_GetHostAccountByExternalNumber",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"account_external",
external_account_number,
)
]),
)
[docs]
async def search_by_internal_number(
self, internal_account_number
) -> list[HostAccountRow]:
return await self.call_hq(
"sdk_GetHostAccountByInternalNumber",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"account_internal",
internal_account_number,
)
]),
)
[docs]
async def update_account_rights(self, parameters: UpdateAccountRightsParameters):
"""
Update the account rights for an account linked to a user. This only works for accounts
explicitly linked to the user, and will not work for accounts linked by a cif.
"""
await self.call_hq(
"sdk_UpdateAccountAccess",
ExecuteStoredProcedure.SqlParameters(parameters.build_sql_parameters()),
)
audit = AuditRecord(self.logger, self.hq_credentials)
await audit.create(
f"{parameters.hostAccountId} access changed to {parameters.calculate_access()} for user id {parameters.userId}",
"SDK Session",
user_id=parameters.userId,
host_account_id=parameters.hostAccountId,
)
[docs]
async def get_customer_host_accounts(
self, customer_ids: list[int]
) -> list[HostAccountCustomerRow]:
"""
Retrieves multiple accounts associated to the customers in the list. For DB performance reason, the customer id list should be limited to 20
:param customer_ids: list of Q2_Customer.CustomerID <= 20
"""
if len(customer_ids) > 20:
raise DatabaseDataError(
"Limit the customer id list to 20 or less for DB performance reasons"
)
# <request><get c="1" /><get c="2" /><get c="3" /></request>
get_customers_list = [E.get(c=str(cust_id)) for cust_id in customer_ids]
get_request_xml = E.request(*get_customers_list)
get_request = tostring(get_request_xml, encoding="utf-8")
params = []
Param(get_request.decode(), D_TYPES.Xml, "request").add_to_param_list(params)
return await self.call_hq(
"sdk_GetCustomerAccountsFromCustomList",
ExecuteStoredProcedure.SqlParameters(params),
)
[docs]
async def get_accounts_by_product(
self, group_id: int, product_id: Optional[int] = None
) -> list[HostAccountCustomerWithProductRow]:
"""
Searches a group of customers for accounts with a specific product id
:param group_id: Group to search [Q2_Group.GroupID]
:param product_id: Product ID of the target product [Q2_Product.ProductID]
"""
param_list = []
Param(group_id, D_TYPES.Int, "group_id").add_to_param_list(param_list)
if product_id:
Param(product_id, D_TYPES.Int, "product_id").add_to_param_list(param_list)
return await self.call_hq(
"sdk_GetCustomersWithProductId",
ExecuteStoredProcedure.SqlParameters(param_list),
)
[docs]
async def change_account_product(
self, account_changes: list[ProductChangeParameters]
) -> ProductChangeResults:
"""
Changes the product of a list of accounts
:param account_changes: list of ProductChangeParameters objects.
:return: ProductChangeResults object
"""
account_nodes = [x.build_xml_node() for x in account_changes]
root = E.Records(*account_nodes)
xml_parameter_string = tostring(root, encoding="utf-8")
param_list = []
Param(xml_parameter_string.decode(), D_TYPES.Xml, "data").add_to_param_list(
param_list
)
results = await self.call_hq(
"bsp_ChangeProduct", ExecuteStoredProcedure.SqlParameters(param_list)
)
result_xml_text = results[0].getchildren()[0].text
result_xml = fromstring(result_xml_text)
result_enum = ProductChangeStatus(result_xml.attrib["status"])
match result_enum:
case ProductChangeStatus.Ok:
result_obj = ProductChangeResults(
result_enum, [x.host_account_id for x in account_changes], []
)
case ProductChangeStatus.AllRecordsSkipped:
result_obj = ProductChangeResults(
result_enum, [], [x.host_account_id for x in account_changes]
)
case ProductChangeStatus.PartialUpdate:
skipped = [int(x.attrib["HostAccountID"]) for x in result_xml.skipped]
success = [
x.host_account_id
for x in account_changes
if x.host_account_id not in skipped
]
result_obj = ProductChangeResults(result_enum, success, skipped)
return result_obj