from lxml.etree import cleanup_namespaces, tostring
from lxml.objectify import E
from q2_sdk.core.dynamic_imports import (
api_ExecuteStoredProcedure as ExecuteStoredProcedure,
)
from q2_sdk.hq.table_row import TableRow
from .db_object import DbObject
[docs]
class TransactionsRow(TableRow):
TransactionID: int
CustomerID: int
UserID: int
CreateDate: str
TransactionType: str
TransactionAmount: float
OriginatingAccountID: int
[docs]
class AchDetailRow(TableRow):
TransactionID: int
CompanyName: str
BatchDiscretionaryData: str
CompanyIdentification: str
CompanyEntryDescription: str
CompanyDescriptiveDate: str
EffectiveEntryDate: str
ABA: str
AccountNumber: str
Amount: float
IdentificationNumber: str
Name: str
DetailDiscretionaryData: str
Addenda: str
EmailAddress: str
AchClassCode: str
RecipientDisplayName: str
ParentID: int
SettlementID: int
HasChildren: bool
AccountType: str
AccountNickName: str
AbaMasked: str
AccountNumberMasked: str
UserRoleID: int
SubsidiaryID: int
AccountTypeDescription: str
IsSameDayAch: bool
RecipientID: int
CustomerID: int
[docs]
class Transactions(DbObject):
REPRESENTATION_ROW_CLASS = TransactionsRow
[docs]
async def get_suspect(self) -> list[TransactionsRow]:
response = await self.call_hq("sdk_GetSuspectTransactions")
return response
[docs]
async def get_ach_details(
self, transaction_ids: list[int], with_recipient_id=False
) -> list[AchDetailRow]:
"""
Takes a list of transaction ids and queries the database for additional ach details
:param: transaction_ids: A list of generated transaction ids that are ach transaction types
:param: with_recipient_id: When set to true, the results include RecipientID
"""
ach_transactions_xml = [E.ach(id=str(x)) for x in transaction_ids]
root = E.root(*ach_transactions_xml)
cleanup_namespaces(root)
request = tostring(root, encoding="utf-8")
parameter = ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Xml, "request", request.decode()
)
if with_recipient_id:
res = await self.call_hq(
"sdk_GetACHTransactionDetails",
ExecuteStoredProcedure.SqlParameters([parameter]),
)
new_res = []
for ele in res:
display_name = str(ele.DisplayName)
ach_display = ele.find("AchName") or display_name[:22]
if (
ele.find("RecipientDisplayName") in (display_name, None)
and ele.Name == ach_display
):
new_res.append(ele)
return new_res if len(new_res) > 0 else res
else:
return await self.call_hq(
"sdk_GetACHTransactionDetailsNoRecipientId",
ExecuteStoredProcedure.SqlParameters([parameter]),
)