from argparse import _SubParsersAction
from functools import partial
from typing import List
from lxml.objectify import IntElement, StringElement
from q2_sdk.core.dynamic_imports import (
api_ExecuteStoredProcedure as ExecuteStoredProcedure,
)
from q2_sdk.core.exceptions import DatabaseDataError
from .customer import Customer
from .db_object import DbObject
from .representation_row_base import RepresentationRowBase
from .user import User
D_TYPES = ExecuteStoredProcedure.DataType
[docs]
class ExternalTransactionRow(RepresentationRowBase):
# object name: type hinting = "column name in the db response"
ExternalTransactionID: IntElement = "ExternalTransactionID"
ExternalTransactionTypeID: IntElement = "ExternalTransactionTypeID"
ExternalTransactionStatusID: IntElement = "ExternalTransactionStatusID"
CreatedDate: StringElement = "CreatedDate"
Description: StringElement = "Description"
CustomerID: IntElement = "CustomerID"
UserID: IntElement = "UserID"
EffectiveDate: StringElement = "EffectiveDate"
TransactionAmount: IntElement = "TransactionAmount"
HostAccountID: IntElement = "HostAccountID"
[docs]
class ExternalTransaction(DbObject):
# GET_BY_NAME_KEY = "column in the db response"
NAME = "ExternalTransaction"
REPRESENTATION_ROW_CLASS = ExternalTransactionRow
[docs]
def add_arguments(self, parser: _SubParsersAction):
subparser = parser.add_parser("get_external_transaction")
subparser.set_defaults(parser="get")
subparser.set_defaults(func=partial(self.get, serialize_for_cli=True))
subparser.add_argument(
"external_transaction_id",
help="Q2_ExternalTransaction.ExternalTransactionID",
)
subparser = parser.add_parser("get_external_transaction_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_external_transaction_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("customer_id", help="Q2_User.UserID")
subparser.add_argument(
"-rc", "--return-count", help="Number of rows to be returned"
)
subparser = parser.add_parser("get_external_transaction_by_host_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, external_transaction_id: int, serialize_for_cli=False
) -> List[ExternalTransactionRow]:
response = await self.call_hq(
"sdk_GetExternalTransaction",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int,
"external_transaction_id",
external_transaction_id,
)
]),
)
if serialize_for_cli:
columns = [
"ExternalTransactionID",
"ExternalTransactionTypeID",
"ExternalTransactionStatusID",
"CreatedDate",
"Description",
"CustomerID",
"UserID",
"EffectiveDate",
"TransactionAmount",
"HostAccountID",
]
response = self.serialize_for_cli(response, columns)
return response
[docs]
async def get_by_customer_id(
self, customer_id, return_count=None, serialize_for_cli=False
):
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_GetExternalTransactionByCustomerId",
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 = [
"ExternalTransactionID",
"ExternalTransactionTypeID",
"ExternalTransactionStatusID",
"CreatedDate",
"Description",
"CustomerID",
"UserID",
"EffectiveDate",
"TransactionAmount",
"HostAccountID",
]
response = self.serialize_for_cli(response, columns)
return response
[docs]
async def get_by_user_id(self, user_id, return_count=None, serialize_for_cli=False):
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_GetExternalTransactionByUserId",
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 = [
"ExternalTransactionID",
"ExternalTransactionTypeID",
"ExternalTransactionStatusID",
"CreatedDate",
"Description",
"CustomerID",
"UserID",
"EffectiveDate",
"TransactionAmount",
"HostAccountID",
]
response = self.serialize_for_cli(response, columns)
return response
[docs]
async def get_by_host_account_id(
self, host_account_id, return_count=None, serialize_for_cli=False
):
response = await self.call_hq(
"sdk_GetExternalTransactionByHostAccountId",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int,
"host_account_id",
host_account_id,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int,
"return_count",
return_count,
),
]),
)
if serialize_for_cli:
columns = [
"ExternalTransactionID",
"ExternalTransactionTypeID",
"ExternalTransactionStatusID",
"CreatedDate",
"Description",
"CustomerID",
"UserID",
"EffectiveDate",
"TransactionAmount",
"HostAccountID",
]
response = self.serialize_for_cli(response, columns)
return response