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