from argparse import _SubParsersAction
from functools import partial
from typing import List, Optional
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 ParamsBuilder
from q2_sdk.hq.table_row import TableRow
from .customer import Customer
from .db_object import DbObject
from .user import User
D_TYPES = ExecuteStoredProcedure.DataType
# Txn-node attributes required by bsp_AddUpdateExternalTransaction
TXN_REQUIRED_FIELDS = (
"TxnType",
"TxnStatus",
"CreatedDate",
"InsertedDate",
"Description",
)
def _non_empty_attrs(node: dict) -> dict:
"""Stringify a node's attributes, dropping empty values."""
return {
key: str(value)
for key, value in node.items()
if value is not None and value != ""
}
[docs]
class ExternalTransactionRow(TableRow):
ExternalTransactionID: int
ExternalTransactionTypeID: int
ExternalTransactionStatusID: int
CreatedDate: str
Description: str
CustomerID: int
UserID: int
EffectiveDate: str
TransactionAmount: int
HostAccountID: int
[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 add_update(
self,
transaction: dict,
wire: Optional[dict] = None,
details: Optional[List[dict]] = None,
) -> List[ExternalTransactionRow]:
"""Add or update an external wire transaction.
Calls the Q2 core proc bsp_AddUpdateExternalTransaction with an
``<ET><Txn><D/><Wire/></Txn></ET>`` XML payload built from the given
dicts (the D and Wire nodes are children of Txn, matching the paths the
proc reads: ``/ET/Txn/D`` and ``//Wire``). The proc upserts on
ExternalTransactionID or ExternalTrackingNumber: include either in
``transaction`` to force an update, omit both to insert.
:param transaction: Txn-node attributes.
Requires TxnType, TxnStatus, CreatedDate, InsertedDate, Description.
:param wire: optional Wire-node attributes (Sender, FromAccount,
Recipient, ReceivingInstitution, ToAccount, ToRouting, BeneFi*,
IntermedFi*, ...). NOTE: when a wire row already exists for the
transaction, the platform proc updates ONLY the Imad field.
:param details: optional list of D-node dicts (ShortName, CreateDate,
DataValue). Omit DataValue to remove a detail element.
:returns: rows of ExternalTransactionID, ExternalTrackingNumber,
updated, inserted
"""
missing = [field for field in TXN_REQUIRED_FIELDS if not transaction.get(field)]
if missing:
raise DatabaseDataError(
f"External transaction is missing required Txn field(s): {', '.join(missing)}"
)
txn_node = E.Txn(**_non_empty_attrs(transaction))
if details:
for detail in details:
txn_node.append(E.D(**_non_empty_attrs(detail)))
if wire:
txn_node.append(E.Wire(**_non_empty_attrs(wire)))
payload = tostring(E.ET(txn_node), encoding="utf-8").decode()
param_list = ParamsBuilder().add_param(D_TYPES.Xml, "data", payload).build()
return await self.call_hq(
"bsp_AddUpdateExternalTransaction",
ExecuteStoredProcedure.SqlParameters(param_list),
)
[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