Source code for q2_sdk.hq.db.generated_transactions

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