Source code for q2_sdk.hq.db.transactions

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]), )