from argparse import _SubParsersAction
from functools import partial
from typing import List
from q2_sdk.core.dynamic_imports import (
api_ExecuteStoredProcedure as ExecuteStoredProcedure,
)
from q2_sdk.core.exceptions import DatabaseDataError
from q2_sdk.hq.table_row import TableRow
from .customer import Customer
from .db_object import DbObject
from .generated_transactions import GeneratedTransactions
from .user import User
[docs]
class FundsTransferRow(TableRow):
TransactionID: int
ToAccountID: int
Principal: float
Interest: float
TransferType: int
IsSameDay: bool
CustomerID: int
UserID: int
CreateDate: str
TransactionStatusID: int
RecurringTransactionID: int
TransactionTypeID: int
OriginatingAccountID: int
TransactionAmount: float
[docs]
class FundsTransfer(DbObject):
# GET_BY_NAME_KEY = "column in the db response"
NAME = "FundsTransfer"
REPRESENTATION_ROW_CLASS = FundsTransferRow
[docs]
def add_arguments(self, parser: _SubParsersAction):
subparser = parser.add_parser("get_funds_transfer")
subparser.set_defaults(parser="get")
subparser.set_defaults(func=partial(self.get, serialize_for_cli=True))
subparser.add_argument(
"transaction_id", help="Q2_GeneratedTransactions.TransferID"
)
subparser = parser.add_parser("get_funds_transfer_by_customer_id")
subparser.set_defaults(parser="get_by_customer_id")
subparser.set_defaults(
func=partial(self.get_by_customer_id, serialize_for_cli=True)
)
subparser.add_argument("customer_id", help="Q2_Customer.CustomerID")
subparser.add_argument(
"-rc", "--return-count", help="Number of rows to be returned"
)
subparser = parser.add_parser("get_funds_transfer_by_user_id")
subparser.set_defaults(parser="get_by_user_id")
subparser.set_defaults(
func=partial(self.get_by_user_id, serialize_for_cli=True)
)
subparser.add_argument("user_id", help="Q2_User.UserID")
subparser.add_argument(
"-rc", "--return-count", help="Number of rows to be returned"
)
subparser = parser.add_parser("get_funds_transfer_by_account_id")
subparser.set_defaults(parser="get_by_host_account_id")
subparser.set_defaults(
func=partial(self.get_by_host_account_id, serialize_for_cli=True)
)
subparser.add_argument("host_account_id", help="Q2_HostAccount.HostAccountID")
subparser.add_argument(
"-rc", "--return-count", help="Number of rows to be returned"
)
[docs]
async def get(
self, transaction_id: int, serialize_for_cli=False
) -> List[FundsTransferRow]:
generated_transaction_obj = GeneratedTransactions(
self.logger, self.hq_credentials, ret_table_obj=True
)
generated_transactions_row = await generated_transaction_obj.get_by_id(
transaction_id
)
if not generated_transactions_row:
raise DatabaseDataError(
f"No GeneratedTransactions with TransactionID {transaction_id} is present"
)
response = await self.call_hq(
"sdk_GetFundsTransfer",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int,
"transaction_id",
transaction_id,
)
]),
)
if serialize_for_cli:
columns = [
"TransactionID",
"ToAccountID",
"Principal",
"Interest",
"TransferType",
"CustomerID",
"CreateDate",
"TransactionStatusID",
"RecurringTransactionID",
"TransactionTypeID",
]
response = self.serialize_for_cli(response, columns)
return response
[docs]
async def get_by_customer_id(
self, customer_id: int, return_count=None, serialize_for_cli=False
) -> List[FundsTransferRow]:
customer_obj = Customer(self.logger, self.hq_credentials, ret_table_obj=True)
customer_row = await customer_obj.get(customer_id=customer_id)
if not customer_row:
raise DatabaseDataError(f"No Customer with CustomerID {customer_id} exists")
response = await self.call_hq(
"sdk_GetFundsTransferByCustomerId",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int, "customer_id", customer_id
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int,
"return_count",
return_count,
),
]),
)
if serialize_for_cli:
columns = [
"TransactionID",
"ToAccountID",
"Principal",
"Interest",
"TransferType",
"CustomerID",
"UserID",
"CreateDate",
"TransactionStatusID",
"RecurringTransactionID",
"TransactionTypeID",
]
response = self.serialize_for_cli(response, columns)
return response
[docs]
async def get_by_user_id(
self, user_id: int, return_count=None, serialize_for_cli=False
) -> List[FundsTransferRow]:
user_obj = User(self.logger, self.hq_credentials, ret_table_obj=True)
customer_row = await user_obj.get(user_id)
if not customer_row:
raise DatabaseDataError(f"No User with UserID {user_id} exists")
response = await self.call_hq(
"sdk_GetFundsTransferByUserId",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int, "user_id", user_id
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int,
"return_count",
return_count,
),
]),
)
if serialize_for_cli:
columns = [
"TransactionID",
"ToAccountID",
"Principal",
"Interest",
"TransferType",
"CustomerID",
"UserID",
"CreateDate",
"TransactionStatusID",
"RecurringTransactionID",
"TransactionTypeID",
]
response = self.serialize_for_cli(response, columns)
return response
[docs]
async def get_by_host_account_id(
self, host_account_id: int, return_count=None, serialize_for_cli=False
) -> List[FundsTransferRow]:
param_list = [
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int,
"host_account_id",
host_account_id,
)
]
if return_count:
assert isinstance(return_count, int), "return_count must be int"
param_list.append(
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int, "return_count", return_count
)
)
response = await self.call_hq(
"sdk_GetFundsTransferByHostAccountId",
ExecuteStoredProcedure.SqlParameters(param_list),
)
if serialize_for_cli:
columns = [
"TransactionID",
"ToAccountID",
"Principal",
"Interest",
"TransferType",
"CustomerID",
"UserID",
"CreateDate",
"TransactionStatusID",
"RecurringTransactionID",
"TransactionTypeID",
]
response = self.serialize_for_cli(response, columns)
return response