from argparse import _SubParsersAction
from dataclasses import dataclass
from datetime import datetime
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 E
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 q2_sdk.hq.table_row import TableRow
from .customer import Customer
from .db_object import DbObject
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(TableRow):
CustomerID: int
TransactionID: int
TransactionStatus: str
TransactionType: str
Description: str
CreatedDate: str
[docs]
class ExternalGeneratedTransfer(TableRow):
TransactionDirection: str
TransactionID: int
CustomerID: int
UserID: int
CreateDate: str
TransactionStatusID: int
TransactionStatus: str
ProcessDate: str
ProcessedDate: str
OriginatingAccountID: int
OriginatingAccountNumberExternal: str
OriginatingABA: str
TransactionAmount: float
AuthorizedUserID: int
AuthorizedDate: str
RecurringTransactionID: int
RecurringMaster: int
Description: str
TargetDescription: str
TargetAccountNumberExternal: str
TargetABA: str
[docs]
class GeneratedTransactionsRow(TableRow):
TransactionID: int
CustomerID: int
CustomerName: str
UserID: int
CreateDate: str
UISourceID: int
UISource: str
TransactionStatusID: int
TransactionStatus: str
TransactionStatusDescription: str
TransactionTypeID: int
TransactionType: str
TransactionTypeDescription: str
ProcessDate: str
ProcessedDate: str
Diagnostic: str
HostAccountID: int
AccountNumberInternal: str
AccountNumberExternal: str
ProductID: int
ProductName: str
ProductTypeID: int
ProductTypeName: str
CurrencyCode: str
AuthorizedByID: int
AuthorizedDate: str
ModifiedByID: int
ModifiedBy: str
ModifiedDate: str
TemplateID: int
HostResult: str
HydraProductTypeCode: str
HostProductTypeCode: str
HydraProductCode: str
HostProductCode: str
ProcessingTypeID: int
ProcessingType: str
ProcessingTypeDescription: str
TargetDescription: str
ParentID: int
SettlementID: int
HasChildren: bool
IsDistributedFundingChild: bool
SkipBalanceCheck: bool
CifInternal: str
AccountNickName: str
EffectiveDate: str
BillPayeeID: int
PayeeDisplayName: str
PayeeName: str
AuditScoreID: int
TranScoreStatus: str
TranScoreWeight: int
LoginScoreID: int
LoginScoreStatus: str
LoginScoreWeight: int
GroupID: int
GroupDesc: str
ToAccountNumberInternal: str
ToProductName: str
ToProductTypeName: str
HasCaseNotes: bool
ScoreAuditID: int
UserRoleID: int
RecipientCount: int
SubsidiaryID: int
GTBatchID: int
RecipientName: str
IntermediaryBankName: str
BeneficiaryBankName: str
MessageToBeneficiary: str
TemplateName: str
TemplateDescription: str
RecipientAccountNumber: str
RecipientAccountType: str
Principal: float
Interest: float
UsdEquivalentAmount: float
RecipientIdentificationNumber: str
IsSameDayAch: bool
CompanyEntryDescription: str
PurposeOfWire: str
AchClassCode: str
HasWageGarnishmentPayments: bool
ReplacesTransactionID: int
ReplacedByTransactionID: int
IsReversed: bool
ReversedGTID: int
IsReversible: bool
ToProductID: int
ToHydraProductTypeCode: str
ToHydraProductCode: str
ToHostProductTypeCode: str
ToHostProductCode: str
IsAchIat: bool
FileName: str
OobAuthRequired: bool
OobAuthProvided: bool
RecurringTransactionID: int
TransactionAmount: str
Description: str
ToHostAccountID: int
ToAccountNumberExternal: str
GtFlavorID: int
[docs]
class GeneratedLoanTransaction(TableRow):
TransactionID: int
CustomerID: int
CustomerName: str
UserID: int
GroupID: int
GroupDesc: str
FirstName: str
LastName: str
CreateDate: str
TransactionStatusID: int
TransactionStatus: str
TransactionTypeID: int
TransactionType: str
ProcessDate: str
ProcessedDate: str
OriginatingAccountID: int
OriginatingAccountNumberInternal: str
OriginatingAccountNumberExternal: str
OriginatingAccountCIFInternal: str
OriginatingAccountCIFExternal: str
OriginatingAccountBranchID: int
TransactionAmount: float
AuthorizedUserID: int
AuthorizedDate: str
RecurringTransactionID: int
RecurringMaster: int
Description: str
TargetDescription: str
ToAccountID: int
TargetAccountNumberInternal: str
TargetAccountNumberExternal: str
TargetAccountCIFInternal: str
TargetAccountCIFExternal: str
TargetAccountBranchID: int
GTDataName: str
GTDataValue: str
Principal: float
Interest: float
[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