from lxml.objectify import IntElement, StringElement, FloatElement, BoolElement, E
from lxml.etree import tostring, cleanup_namespaces
from q2_sdk.core.dynamic_imports import (
api_ExecuteStoredProcedure as ExecuteStoredProcedure,
)
from .db_object import DbObject
from .representation_row_base import RepresentationRowBase
[docs]
class TransactionsRow(RepresentationRowBase):
TransactionID: IntElement = "TransactionID"
CustomerID: IntElement = "CustomerID"
UserID: IntElement = "UserID"
CreateDate: StringElement = "CreateDate"
TransactionType: StringElement = "ShortName"
TransactionAmount: FloatElement = "TransactionAmount"
OriginatingAccountID: IntElement = "OriginationAccountID"
[docs]
class AchDetailRow(RepresentationRowBase):
TransactionID: IntElement = "TransactionID"
CompanyName: StringElement = "CompanyName"
BatchDiscretionaryData: StringElement = "BatchDiscretionaryData"
CompanyIdentification: StringElement = "CompanyIdentification"
CompanyEntryDescription: StringElement = "CompanyEntryDescription"
CompanyDescriptiveDate: StringElement = "CompanyDescriptiveDate"
EffectiveEntryDate: StringElement = "EffectiveEntryDate"
ABA: StringElement = "ABA"
AccountNumber: StringElement = "AccountNumber"
Amount: FloatElement = "Amount"
IdentificationNumber: StringElement = "IdentificationNumber"
Name: StringElement = "Name"
DetailDiscretionaryData: StringElement = "DetailDiscretionaryData"
Addenda: StringElement = "Addenda"
EmailAddress: StringElement = "EmailAddress"
AchClassCode: StringElement = "AchClassCode"
RecipientDisplayName: StringElement = "RecipientDisplayName"
ParentID: IntElement = "ParentID"
SettlementID: IntElement = "SettlementID"
HasChildren: BoolElement = "HasChildren"
AccountType: StringElement = "AccountType"
AccountNickName: StringElement = "AccountNickName"
AbaMasked: StringElement = "AbaMasked"
AccountNumberMasked: StringElement = "AccountNumberMasked"
UserRoleID: IntElement = "UserRoleID"
SubsidiaryID: IntElement = "SubsidiaryID"
AccountTypeDescription: StringElement = "AccountTypeDescription"
IsSameDayAch: BoolElement = "IsSameDayAch"
RecipientID: IntElement = "RecipientID"
CustomerID: IntElement = "CustomerID"
[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]),
)