Source code for q2_sdk.hq.db.generated_transactions

from datetime import datetime
from argparse import _SubParsersAction
from dataclasses import dataclass
from enum import Enum
from functools import partial
from typing import Optional

from dateutil import parser as date_parser
from lxml.etree import tostring
from lxml.objectify import BoolElement, E, FloatElement, IntElement, StringElement

from q2_sdk.core.dynamic_imports import (
    api_ExecuteStoredProcedure as ExecuteStoredProcedure,
)
from q2_sdk.core.exceptions import DatabaseDataError
from q2_sdk.hq.models.hq_params.stored_procedure import Param
from q2_sdk.hq.models.transaction_info import TransactionStatus

from .customer import Customer
from .db_object import DbObject
from .representation_row_base import RepresentationRowBase
from .user_logon import UserLogon

D_TYPES = ExecuteStoredProcedure.DataType


[docs] @dataclass class GetGeneratedTransactionsParams: customer_id: int from_date: str to_date: str recurring_transaction_id: Optional[int] = None
[docs] @dataclass class AddTransactionNoteParams: transaction_id: int admin_login_name: int note: int note_date: Optional[str] = None
[docs] class DateColumn(Enum): CreateDate = "Create" ProcessDate = "Process"
[docs] class DateTimeEntryError(Exception): """Transaction CreateDates are not entered correctly"""
[docs] class GeneratedTransactionsPending(RepresentationRowBase): CustomerID: IntElement = "CustomerID" TransactionID: IntElement = "TransactionID" TransactionStatus: StringElement = "TransactionStatus" TransactionType: StringElement = "TransactionType" Description: StringElement = "Description" CreatedDate: StringElement = "CreatedDate"
[docs] class ExternalGeneratedTransfer(RepresentationRowBase): TransactionDirection: StringElement = "TransactionDirection" TransactionID: IntElement = "TransactionID" CustomerID: IntElement = "CustomerID" UserID: IntElement = "UserID" CreateDate: StringElement = "CreateDate" TransactionStatusID: IntElement = "TransactionStatusID" TransactionStatus: StringElement = "TransactionStatus" ProcessDate: StringElement = "ProcessDate" ProcessedDate: StringElement = "ProcessedDate" OriginatingAccountID: IntElement = "OriginatingAccountID" OriginatingAccountNumberExternal: StringElement = "OriginatingAccountNumberExternal" OriginatingABA: StringElement = "OriginatingABA" TransactionAmount: FloatElement = "TransactionAmount" AuthorizedUserID: IntElement = "AuthorizedUserID" AuthorizedDate: StringElement = "AuthorizedDate" RecurringTransactionID: IntElement = "RecurringTransactionID" RecurringMaster: IntElement = "RecurringMaster" Description: StringElement = "Description" TargetDescription: StringElement = "TargetDescription" TargetAccountNumberExternal: StringElement = "TargetAccountNumberExternal" TargetABA: StringElement = "TargetABA"
[docs] class GeneratedTransactionsRow(RepresentationRowBase): # object name: type hinting = "column name in the db response" TransactionID: IntElement = "TransactionID" CustomerID: IntElement = "CustomerID" CustomerName: StringElement = "CustomerName" UserID: IntElement = "UserID" CreateDate: StringElement = "CreateDate" UISourceID: IntElement = "UISourceID" UISource: StringElement = "UISource" TransactionStatusID: IntElement = "TransactionStatusID" TransactionStatus: StringElement = "TransactionStatus" TransactionStatusDescription: StringElement = "TransactionStatusDescription" TransactionTypeID: IntElement = "TransactionTypeID" TransactionType: StringElement = "TransactionType" TransactionTypeDescription: StringElement = "TransactionTypeDescription" ProcessDate: StringElement = "ProcessDate" ProcessedDate: StringElement = "ProcessedDate" Diagnostic: StringElement = "Diagnostic" HostAccountID: IntElement = "HostAccountID" AccountNumberInternal: StringElement = "AccountNumberInternal" AccountNumberExternal: StringElement = "AccountNumberExternal" ProductID: IntElement = "ProductID" ProductName: StringElement = "ProductName" ProductTypeID: IntElement = "ProductTypeID" ProductTypeName: StringElement = "ProductTypeName" CurrencyCode: StringElement = "CurrencyCode" AuthorizedByID: IntElement = "AuthorizedByID" AuthorizedDate: StringElement = "AuthorizedDate" ModifiedByID: IntElement = "ModifiedByID" ModifiedBy: StringElement = "ModifiedBy" ModifiedDate: StringElement = "ModifiedDate" TemplateID: IntElement = "TemplateID" HostResult: StringElement = "HostResult" HydraProductTypeCode: StringElement = "HydraProductTypeCode" HostProductTypeCode: StringElement = "HostProductTypeCode" HydraProductCode: StringElement = "HydraProductCode" HostProductCode: StringElement = "HostProductCode" ProcessingTypeID: IntElement = "ProcessingTypeID" ProcessingType: StringElement = "ProcessingType" ProcessingTypeDescription: StringElement = "ProcessingTypeDescription" TargetDescription: StringElement = "TargetDescription" ParentID: IntElement = "ParentID" SettlementID: IntElement = "SettlementID" HasChildren: BoolElement = "HasChildren" IsDistributedFundingChild: BoolElement = "IsDistributedFundingChild" SkipBalanceCheck: BoolElement = "SkipBalanceCheck" CifInternal: StringElement = "CifInternal" AccountNickName: StringElement = "AccountNickName" EffectiveDate: StringElement = "EffectiveDate" BillPayeeID: IntElement = "BillPayeeID" PayeeDisplayName: StringElement = "PayeeDisplayName" PayeeName: StringElement = "PayeeName" AuditScoreID: IntElement = "AuditScoreID" TranScoreStatus: StringElement = "TranScoreStatus" TranScoreWeight: IntElement = "TranScoreWeight" LoginScoreID: IntElement = "LoginScoreID" LoginScoreStatus: StringElement = "LoginScoreStatus" LoginScoreWeight: IntElement = "LoginScoreWeight" GroupID: IntElement = "GroupID" GroupDesc: StringElement = "GroupDesc" ToAccountNumberInternal: StringElement = "ToAccountNumberInternal" ToProductName: StringElement = "ToProductName" ToProductTypeName: StringElement = "ToProductTypeName" HasCaseNotes: BoolElement = "HasCaseNotes" ScoreAuditID: IntElement = "ScoreAuditID" UserRoleID: IntElement = "UserRoleID" RecipientCount: IntElement = "RecipientCount" SubsidiaryID: IntElement = "SubsidiaryID" GTBatchID: IntElement = "GTBatchID" RecipientName: StringElement = "RecipientName" IntermediaryBankName: StringElement = "IntermediaryBankName" BeneficiaryBankName: StringElement = "BeneficiaryBankName" MessageToBeneficiary: StringElement = "MessageToBeneficiary" TemplateName: StringElement = "TemplateName" TemplateDescription: StringElement = "TemplateDescription" RecipientAccountNumber: StringElement = "RecipientAccountNumber" RecipientAccountType: StringElement = "RecipientAccountType" Principal: FloatElement = "Principal" Interest: FloatElement = "Interest" UsdEquivalentAmount: FloatElement = "UsdEquivalentAmount" RecipientIdentificationNumber: StringElement = "RecipientIdentificationNumber" IsSameDayAch: BoolElement = "IsSameDayAch" CompanyEntryDescription: StringElement = "CompanyEntryDescription" PurposeOfWire: StringElement = "PurposeOfWire" AchClassCode: StringElement = "AchClassCode" HasWageGarnishmentPayments: BoolElement = "HasWageGarnishmentPayments" ReplacesTransactionID: IntElement = "ReplacesTransactionID" ReplacedByTransactionID: IntElement = "ReplacedByTransactionID" IsReversed: BoolElement = "IsReversed" ReversedGTID: IntElement = "ReversedGTID" IsReversible: BoolElement = "IsReversible" ToProductID: IntElement = "ToProductID" ToHydraProductTypeCode: StringElement = "ToHydraProductTypeCode" ToHydraProductCode: StringElement = "ToHydraProductCode" ToHostProductTypeCode: StringElement = "ToHostProductTypeCode" ToHostProductCode: StringElement = "ToHostProductCode" IsAchIat: BoolElement = "IsAchIat" FileName: StringElement = "FileName" OobAuthRequired: BoolElement = "OobAuthRequired" OobAuthProvided: BoolElement = "OobAuthProvided" RecurringTransactionID: IntElement = "RecurringTransactionID" TransactionAmount: StringElement = "TransactionAmount" Description: StringElement = "Description" ToHostAccountID: IntElement = "ToHostAccountID" ToAccountNumberExternal: StringElement = "ToAccountNumberExternal" GtFlavorID: IntElement = "GtFlavorID"
[docs] class GeneratedLoanTransaction(RepresentationRowBase): TransactionID: IntElement = "TransactionID" CustomerID: IntElement = "CustomerID" CustomerName: StringElement = "CustomerName" UserID: IntElement = "UserID" GroupID: IntElement = "GroupID" GroupDesc: StringElement = "GroupDesc" FirstName: StringElement = "FirstName" LastName: StringElement = "LastName" CreateDate: StringElement = "CreateDate" TransactionStatusID: IntElement = "TransactionStatusID" TransactionStatus: StringElement = "TransactionStatus" TransactionTypeID: IntElement = "TransactionTypeID" TransactionType: StringElement = "TransactionType" ProcessDate: StringElement = "ProcessDate" ProcessedDate: StringElement = "ProcessedDate" OriginatingAccountID: IntElement = "OriginatingAccountID" OriginatingAccountNumberInternal: StringElement = "OriginatingAccountNumberInternal" OriginatingAccountNumberExternal: StringElement = "OriginatingAccountNumberExternal" OriginatingAccountCIFInternal: StringElement = "OriginatingAccountCIFInternal" OriginatingAccountCIFExternal: StringElement = "OriginatingAccountCIFExternal" OriginatingAccountBranchID: IntElement = "OriginatingAccountBranchID" TransactionAmount: FloatElement = "TransactionAmount" AuthorizedUserID: IntElement = "AuthorizedUserID" AuthorizedDate: StringElement = "AuthorizedDate" RecurringTransactionID: IntElement = "RecurringTransactionID" RecurringMaster: IntElement = "RecurringMaster" Description: StringElement = "Description" TargetDescription: StringElement = "TargetDescription" ToAccountID: IntElement = "ToAccountID" TargetAccountNumberInternal: StringElement = "TargetAccountNumberInternal" TargetAccountNumberExternal: StringElement = "TargetAccountNumberExternal" TargetAccountCIFInternal: StringElement = "TargetAccountCIFInternal" TargetAccountCIFExternal: StringElement = "TargetAccountCIFExternal" TargetAccountBranchID: IntElement = "TargetAccountBranchID" GTDataName: StringElement = "GTDataName" GTDataValue: StringElement = "GTDataValue" Principal: FloatElement = "Principal" Interest: FloatElement = "Interest"
[docs] class GeneratedTransactions(DbObject): # GET_BY_NAME_KEY = "column in the db response" NAME = "GeneratedTransactions" REPRESENTATION_ROW_CLASS = GeneratedTransactionsRow
[docs] def add_arguments(self, parser: _SubParsersAction): subparser = parser.add_parser("get_generated_transactions") subparser.set_defaults(parser="get") subparser.set_defaults(func=partial(self.get, serialize_for_cli=True)) subparser.add_argument("customer_id", help="Q2_Customer.CustomerID", type=int) subparser.add_argument( "from_date", help='Date from which we want to see the transaction data. Ex:"2021-08-26 10:40:49.000" ' "Q2_GeneratedTransactions.CreateDate", type=str, ) subparser.add_argument( "to_date", help="Date to which we want to see the transaction data. " "Q2_GeneratedTransactions.CreateDate", type=str, ) subparser.add_argument( "--recurring_transaction_id", help="Q2_RecurringTransactionID.RecurringTransactionID", type=int, ) subparser = parser.add_parser("get_generated_transactions_by_id") subparser.set_defaults(parser="get_by_id") subparser.set_defaults(func=partial(self.get_by_id, serialize_for_cli=True)) subparser.add_argument( "transaction_id", help="Q2_GeneratedTransactions.TransactionID", type=int ) subparser = parser.add_parser("get_generated_transactions_by_recurring_id") subparser.set_defaults(parser="get_by_recurring_id") subparser.set_defaults( func=partial(self.get_by_recurring_id, serialize_for_cli=True) ) subparser.add_argument( "recurring_transaction_id", help="Q2_RecurringTransaction.RecurringTransactionID", type=int, )
[docs] async def get( self, customer_id: int, from_date: str, to_date: str, recurring_transaction_id=None, serialize_for_cli=False, ) -> list[GeneratedTransactionsRow]: customer = Customer(self.logger, hq_credentials=self.hq_credentials) customer_row = await customer.get(customer_id=int(customer_id)) if not customer_row: raise DatabaseDataError(f"Customer with ID {customer_id} does not exist") from_date_datetime = date_parser.parse(from_date) to_date_datetime = date_parser.parse(to_date) if from_date_datetime > to_date_datetime: raise DateTimeEntryError( f"Transaction from_date {from_date} should be before transaction to_date {to_date}" ) sql_params = self._build_get_parameters( GetGeneratedTransactionsParams( customer_id, from_date, to_date, recurring_transaction_id ) ) response = await self.call_hq( "sdk_GetGeneratedTransactions", ExecuteStoredProcedure.SqlParameters(sql_params), ) if serialize_for_cli: columns = [ "TransactionID", "CustomerID", "UserID", "CreateDate", "UISourceID", "TransactionStatusID", "TransactionStatus", "TransactionTypeID", "RecurringTransactionID", ] response = self.serialize_for_cli(response, columns) return response
def _build_get_parameters(self, get_transaction: GetGeneratedTransactionsParams): parameters = [] possible_params = [ Param(get_transaction.customer_id, D_TYPES.Int, "customer_id"), Param( get_transaction.from_date, D_TYPES.DateTime, "from_transaction_create_date", ), Param( get_transaction.to_date, D_TYPES.DateTime, "to_transaction_create_date" ), Param( get_transaction.recurring_transaction_id, D_TYPES.Int, "recurring_transaction_id", ), ] for item in possible_params: item.add_to_param_list(parameters) return parameters
[docs] async def get_by_id( self, transaction_id, serialize_for_cli=False ) -> list[GeneratedTransactionsRow]: parameters = [] Param(transaction_id, D_TYPES.Int, "transaction_id").add_to_param_list( parameters ) response = await self.call_hq( "sdk_GetGeneratedTransactionsById", sql_parameters=ExecuteStoredProcedure.SqlParameters(parameters), ) if serialize_for_cli: columns = [ "TransactionID", "CustomerID", "UserID", "CreateDate", "UISourceID", "TransactionStatusID", "TransactionStatus", "TransactionTypeID", ] response = self.serialize_for_cli(response, columns) return response
[docs] async def get_by_recurring_id( self, recurring_transaction_id, serialize_for_cli=False ) -> list[GeneratedTransactionsRow]: parameters = [] Param( recurring_transaction_id, D_TYPES.Int, "recurring_transaction_id" ).add_to_param_list(parameters) response = await self.call_hq( "sdk_GetGeneratedTransactionsByRecurringId", sql_parameters=ExecuteStoredProcedure.SqlParameters(parameters), ) if serialize_for_cli: columns = [ "TransactionID", "CustomerID", "UserID", "CreateDate", "UISourceID", "TransactionStatusID", "TransactionStatus", "TransactionTypeID", "RecurringTransactionID", ] response = self.serialize_for_cli(response, columns) return response
[docs] async def add_transaction_note( self, transaction_id, admin_login_name, note, note_date: Optional[str] = None ): """ Adds a note to transaction :param transaction_id: The id of the transaction the note is added to :param admin_login_name: The CSR login name who is adding the note :param note: the note contents :param note_date: Datetime in the form of YYYY-MM-DD HH:MM:SS. If not provided, a "now" timestamp will be used """ assert isinstance(transaction_id, int), "Please provide a valid transaction ID" transaction_exist = await self.get_by_id(transaction_id) print(transaction_exist) if not transaction_exist: raise DatabaseDataError( f"Transaction with ID {transaction_id} does not exist" ) admin_login_check = UserLogon(self.logger, hq_credentials=self.hq_credentials) admin_login = await admin_login_check.get( logon_name=admin_login_name, admin=True ) if not admin_login: raise DatabaseDataError(f"Admin login {admin_login_name} does not exist") login_id = admin_login[0].AdminUserLogonID.text time_stamp_format = "%Y-%m-%dT%H:%M:%S" time_stamp = datetime.now().strftime(time_stamp_format) if note_date: try: time_stamp = date_parser.parse(note_date).strftime(time_stamp_format) except ValueError as date_error: raise DateTimeEntryError( "node_date format is incorrect" ) from date_error transaction_params = AddTransactionNoteParams( transaction_id, admin_login_name, note, note_date ) parameters = self._build_add_parameters( transaction_params, time_stamp, login_id ) await self.call_hq( "Q2_AddCaseAnalysisNote", sql_parameters=ExecuteStoredProcedure.SqlParameters(parameters), )
def _build_add_parameters( self, add_transaction: AddTransactionNoteParams, timestamp, login_id ): parameters = [] possible_params = [ Param(add_transaction.transaction_id, D_TYPES.Int, "transactionID"), Param(timestamp, D_TYPES.DateTime, "noteDate"), Param(login_id, D_TYPES.Int, "adminUserLoginID"), Param(add_transaction.note, D_TYPES.VarChar, "caseNote"), ] for item in possible_params: item.add_to_param_list(parameters) return parameters
[docs] async def get_by_create_date_range( self, start_date: datetime, end_date: datetime, page_number=1, page_size=200, transaction_status_filter: Optional[TransactionStatus] = None, ) -> list[ExternalGeneratedTransfer]: """ Get external transactions by create date range and supports pagination :param start_date: The start date of the date range, as a datetime object :param end_date: The end date of the date range, as a datetime object :param page_number: the starting point for pagination. Defaults to 1 :param page_size: The number of transactions to get per page :param transaction_status_filter: an optional filter to get transactions that are in a specific status :return: A list of transactions that fit the provided parameters """ offset = (page_number - 1) * page_size assert offset >= 0, "page_number must be 1 or greater" sql_params = [] Param( start_date.isoformat(), D_TYPES.DateTime, "FromCreateDate" ).add_to_param_list(sql_params) Param(end_date.isoformat(), D_TYPES.DateTime, "ToCreateDate").add_to_param_list( sql_params ) Param(offset, D_TYPES.Int, "Offset").add_to_param_list(sql_params) Param(page_size, D_TYPES.Int, "ReturnCount").add_to_param_list(sql_params) if transaction_status_filter: Param( transaction_status_filter.value, D_TYPES.VarChar, "OptionalStatusFilter" ).add_to_param_list(sql_params) response = await self.call_hq( "sdk_GetExternalTransfersByDate", ExecuteStoredProcedure.SqlParameters(sql_params), ) return response
[docs] async def get_by_process_date_range( self, start_date: datetime, end_date: datetime, page_number=1, page_size=200, transaction_status_filter: Optional[TransactionStatus] = None, ) -> list[ExternalGeneratedTransfer]: """ Get external transactions by process date range and supports pagination :param start_date: The start date of the date range, as a datetime object :param end_date: The end date of the date range, as a datetime object :param page_number: the starting point for pagination. Defaults to 1 :param page_size: The number of transactions to get per page :param transaction_status_filter: an optional filter to get transactions that are in a specific status :return: A list of transactions that fit the provided parameters """ offset = (page_number - 1) * page_size assert offset >= 0, "page_number must be 1 or greater" sql_params = [] Param( start_date.isoformat(), D_TYPES.DateTime, "FromProcessDate" ).add_to_param_list(sql_params) Param( end_date.isoformat(), D_TYPES.DateTime, "ToProcessDate" ).add_to_param_list(sql_params) Param(offset, D_TYPES.Int, "Offset").add_to_param_list(sql_params) Param(page_size, D_TYPES.Int, "ReturnCount").add_to_param_list(sql_params) if transaction_status_filter: Param( transaction_status_filter.value, D_TYPES.VarChar, "OptionalStatusFilter" ).add_to_param_list(sql_params) response = await self.call_hq( "sdk_GetExternalTransfersByProcessDate", ExecuteStoredProcedure.SqlParameters(sql_params), ) return response
[docs] async def get_loan_transactions_by_date_range( self, start_date: datetime, end_date: datetime, date_column: DateColumn, page_number=1, page_size=200, transaction_status_filter: Optional[TransactionStatus] = None, filter_on_gt_data_name: Optional[str] = None, ) -> list[GeneratedLoanTransaction]: """ Get generated transactions to hydra product type L by date range and supports pagination :param start_date: The start date of the date range, as a datetime object :param end_date: The end date of the date range, as a datetime object :param date_column: The date column to filter the date on :param page_number: the starting point for pagination. Defaults to 1 :param page_size: The number of transactions to get per page :param transaction_status_filter: an optional filter to get transactions that are in a specific status :param filter_on_gt_data_name: an optional filter to only get transactions with specific gt data attached :return: A list of transactions that fit the provided parameters """ offset = (page_number - 1) * page_size assert offset >= 0, "page_number must be 1 or greater" sql_params = [] Param(start_date.isoformat(), D_TYPES.DateTime, "FromDate").add_to_param_list( sql_params ) Param(end_date.isoformat(), D_TYPES.DateTime, "ToDate").add_to_param_list( sql_params ) Param(date_column.value, D_TYPES.VarChar, "DateColumnToUse").add_to_param_list( sql_params ) Param(offset, D_TYPES.Int, "Offset").add_to_param_list(sql_params) Param(page_size, D_TYPES.Int, "ReturnCount").add_to_param_list(sql_params) if filter_on_gt_data_name: Param( filter_on_gt_data_name, D_TYPES.VarChar, "FilterOnGtDataName" ).add_to_param_list(sql_params) if transaction_status_filter: Param( transaction_status_filter.value, D_TYPES.VarChar, "OptionalStatusFilter" ).add_to_param_list(sql_params) response = await self.call_hq( "sdk_GetLoanTransfersByDate", ExecuteStoredProcedure.SqlParameters(sql_params), ) return response
[docs] async def get_transactions_by_date_range_with_gt_data( self, start_date: datetime, end_date: datetime, date_column: DateColumn, page_number=1, page_size=200, transaction_status_filter: Optional[TransactionStatus] = None, filter_on_gt_data_name: Optional[str] = None, customer_id: Optional[int] = None, ) -> list[GeneratedLoanTransaction]: """ Get generated transactions with gt data by date range and supports pagination :param start_date: The start date of the date range, as a datetime object :param end_date: The end date of the date range, as a datetime object :param date_column: The date column to filter the date on :param page_number: the starting point for pagination. Defaults to 1 :param page_size: The number of transactions to get per page :param transaction_status_filter: an optional filter to get transactions that are in a specific status :param filter_on_gt_data_name: an optional filter to only get transactions with specific gt data attached :param customer_id: If provided, will filter the results by Q2_Customer.CustomerID :return: A list of transactions that fit the provided parameters """ offset = (page_number - 1) * page_size assert offset >= 0, "page_number must be 1 or greater" sql_params = [] Param(start_date.isoformat(), D_TYPES.DateTime, "FromDate").add_to_param_list( sql_params ) Param(end_date.isoformat(), D_TYPES.DateTime, "ToDate").add_to_param_list( sql_params ) Param(offset, D_TYPES.Int, "Offset").add_to_param_list(sql_params) Param(page_size, D_TYPES.Int, "ReturnCount").add_to_param_list(sql_params) if filter_on_gt_data_name: Param( filter_on_gt_data_name, D_TYPES.VarChar, "FilterOnGtDataName" ).add_to_param_list(sql_params) if transaction_status_filter: Param( transaction_status_filter.value, D_TYPES.VarChar, "OptionalStatusFilter" ).add_to_param_list(sql_params) if customer_id: Param(customer_id, D_TYPES.Int, "CustomerID").add_to_param_list(sql_params) match date_column: case DateColumn.CreateDate: if customer_id: stored_proc_name = "sdk_GetTransactionsWithGTDataCreateDateByCustID" else: stored_proc_name = "sdk_GetTransactionsWithGTDataCreateDate" case DateColumn.ProcessDate: if customer_id: stored_proc_name = ( "sdk_GetTransactionsWithGTDataProcessDateByCustID" ) else: stored_proc_name = "sdk_GetTransactionsWithGTDataProcessDate" case _: stored_proc_name = None if not stored_proc_name: raise DatabaseDataError("Date Column not supported") response = await self.call_hq( stored_proc_name, ExecuteStoredProcedure.SqlParameters(sql_params) ) return response
[docs] async def get_pending_transactions( self, page_number: int = 1, page_size: int = 200, transaction_type_ids_to_omit: Optional[list[int]] = None, ) -> list[GeneratedTransactionsPending]: """ Get pending transactions. Pending transactions include the following statuses: PendingProcessing, Authorized, Drafted, Suspect :param: page_number: the starting point for pagination. Defaults to 1 :param: page_size: The number of transactions to get per page. Defaults to 200 :param: transaction_type_ids_to_omit: The transaction types to omit from the search """ assert page_number > 0, "page_number must be 1 or greater" ids_to_ignore = [] if transaction_type_ids_to_omit: assert isinstance(transaction_type_ids_to_omit, list), ( "omit_transaction_type_ids must be a list" ) ids_to_ignore = [ E.id(str(omit_id)) for omit_id in transaction_type_ids_to_omit ] offset = (page_number - 1) * page_size #'<TransactionTypesToOmit><id>1</id><id>2</id><id>3</id></TransactionTypesToOmit>' transaction_types_xml = tostring(E.TransactionTypesToOmit(*ids_to_ignore)) sql_params = [] Param(offset, D_TYPES.Int, "Offset").add_to_param_list(sql_params) Param(page_size, D_TYPES.Int, "ReturnCount").add_to_param_list(sql_params) Param( transaction_types_xml, D_TYPES.Xml, "OmitTransactionTypes" ).add_to_param_list(sql_params) response = await self.call_hq( "sdk_GetPendingTransactions", ExecuteStoredProcedure.SqlParameters(sql_params), use_json=False, ) return response