import asyncio
from argparse import _SubParsersAction
from dataclasses import dataclass
from functools import partial
from typing import Optional, List
from lxml.objectify import ObjectifiedElement
from q2_sdk.core.dynamic_imports import (
api_ExecuteStoredProcedure as ExecuteStoredProcedure,
)
from .db_object import DbObject
from q2_sdk.hq.table_row import TableRow
from q2_sdk.core.cli import textui, colored
IS_ENABLED_DEFAULT = True
IS_ACTIVE_MODE_DEFAULT = False
# DBObject Stored Procs
GET_TIP_CONFIG_STORED_PROC = "tip_GetConfig"
GET_TIP_CONFIG_BY_GT_TYPE = "tip_GetTipConfigByGTTypeShortName"
ADD_DEFAULT_TIP_CONFIG_STORED_PROC = "tip_AddDefaultTipConfig"
# GT Payload Stored Procs
FUNDS_TRANSFER_STORED_PROC = "tip_GetRequestPayloadFundsTransfer"
INTERNATIONAL_WIRE_STORED_PROC = "tip_GetRequestPayloadInternationalWire"
DOMESTIC_WIRE_STORED_PROC = "tip_GetRequestPayloadDomesticWire"
ACH_STORED_PROC = "tip_GetRequestPayloadAch"
ACH_PASS_THRU_STORED_PROC = "tip_GetRequestPayloadAchPassThru"
EXTERNAL_TRANSFER_STORED_PROC = "tip_GetRequestPayloadExternalTransfer"
GT_SHORT_NAME_TO_STORED_PROC_MAP = {
"FundsTransfer": FUNDS_TRANSFER_STORED_PROC,
"InternationalWire": INTERNATIONAL_WIRE_STORED_PROC,
"DomesticWire": DOMESTIC_WIRE_STORED_PROC,
"ExternalTransfer": EXTERNAL_TRANSFER_STORED_PROC,
"AchPayment": ACH_STORED_PROC,
"AchCollection": ACH_STORED_PROC,
"AchBatch": ACH_STORED_PROC,
"Payroll": ACH_STORED_PROC,
"AchPassThru": ACH_PASS_THRU_STORED_PROC,
}
[docs]
class GetTipConfigRow(TableRow):
EndpointURL: str
PayloadStoredProcName: str
TimeoutInMs: int
OptionalConfigJson: str
IsEnabled: bool
IsActiveMode: bool
FallbackTransactionStatusShortName: str
[docs]
class GetTipConfigByGtTypeRow(TableRow):
ConfigID: int
OptionalGtTypeShortName_: str
OptionalUiSourceShortName_: str
OptionalGtFlavorShortName_: str
OptionalGroupID_: int
IsEnabled: bool
IsActiveMode: bool
FallbackTransactionStatusShortName_: str
EndpointURL: str
PayloadStoredProcName: str
TimeoutInMs: int
OptionalConfigJson: str
[docs]
@dataclass
class TipConfigDefaultParamsFactory:
endpoint_url: str
optional_ui_source_short_name: Optional[str] = None
optional_gt_flavor_short_name: Optional[str] = None
optional_group_id: Optional[int] = None
is_enabled: bool = IS_ENABLED_DEFAULT
is_active_mode: bool = IS_ACTIVE_MODE_DEFAULT
fallback_transaction_status_short_name: str = "Drafted"
[docs]
def build_sql_params(
self,
gt_type_short_name: str,
) -> ExecuteStoredProcedure.SqlParameters:
return ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"GtTypeShortName",
gt_type_short_name,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"OptionalUiSourceShortName",
self.optional_ui_source_short_name,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"OptionalGtFlavorShortName",
self.optional_gt_flavor_short_name,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int,
"OptionalGroupID",
self.optional_group_id,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Bit,
"IsEnabled",
self.is_enabled,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Bit,
"IsActiveMode",
self.is_active_mode,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"FallbackTransactionStatusShortName",
self.fallback_transaction_status_short_name,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"EndpointURL",
self.endpoint_url,
),
# should we create a system data type ?
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"PayloadStoredProcName",
GT_SHORT_NAME_TO_STORED_PROC_MAP[gt_type_short_name],
),
])
[docs]
class TipConfig(DbObject):
NAME = "TipConfig"
REPRESENTATION_ROW_CLASS = GetTipConfigRow
[docs]
def add_arguments(self, parser: _SubParsersAction):
subparser = parser.add_parser("get_tip_config")
subparser.set_defaults(parser="get")
subparser.set_defaults(func=partial(self.get, serialize_for_cli=True))
subparser.add_argument("--user_id", type=int, help="Q2_User.UserID")
subparser.add_argument(
"--gt_type_short_name",
type=str,
help="Q2_GeneratedTransactionTypes.ShortName",
)
subparser.add_argument(
"--ui_source_short_name",
type=str,
help="Q2_UISource.ShortName",
)
subparser.add_argument(
"--optional_gt_flavor_short_name",
type=str,
help="Q2_GTFlavor.ShortName",
)
subparser = parser.add_parser("get_tip_config_by_gt_type")
subparser.set_defaults(parser="get_by_gt_type")
subparser.set_defaults(
func=partial(self.get_by_gt_type, serialize_for_cli=True)
)
subparser.add_argument(
"--gt_type_short_name",
type=str,
help="Q2_GeneratedTransactionTypes.ShortName",
)
[docs]
async def get(
self,
user_id: int,
gt_type_short_name: str,
ui_source_short_name: Optional[str] = None,
optional_gt_flavor_short_name: Optional[str] = None,
serialize_for_cli: bool = False,
) -> List[GetTipConfigRow]:
assert isinstance(user_id, int), "Please supply a valid user_id"
assert isinstance(gt_type_short_name, str), (
"Please supply a valid gt_type_short_name"
)
response = await self.call_hq(
GET_TIP_CONFIG_STORED_PROC,
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int, "UserID", user_id
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"GtTypeShortName",
gt_type_short_name,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"UiSourceShortName",
ui_source_short_name or "NULL",
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"OptionalGtFlavorShortName",
optional_gt_flavor_short_name or "NULL",
),
]),
)
if serialize_for_cli:
columns = [
"EndpointURL",
"PayloadStoredProcName",
"IsEnabled",
"IsActiveMode",
"TimeoutInMs",
"OptionalConfigJson",
"FallbackTransactionStatusShortName",
]
response = self.serialize_for_cli(response, columns)
return response
[docs]
async def get_by_gt_type(
self,
gt_type_short_name: str,
serialize_for_cli: bool = False,
) -> List[GetTipConfigByGtTypeRow]:
assert isinstance(gt_type_short_name, str), (
"Please supply a valid gt_type_short_name"
)
response = await self.call_hq(
GET_TIP_CONFIG_BY_GT_TYPE,
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"GtTypeShortName",
gt_type_short_name,
),
]),
)
if serialize_for_cli:
columns = [
"ConfigID",
"OptionalGtTypeShortName_",
"OptionalUiSourceShortName_",
"OptionalGtFlavorShortName_",
"OptionalGroupID_",
"IsEnabled",
"IsActiveMode",
"FallbackTransactionStatusShortName_",
"EndpointURL",
"PayloadStoredProcName",
"TimeoutInMs",
"OptionalConfigJson",
]
response = self.serialize_for_cli(response, columns)
return response
[docs]
async def add_default_record_for_gt_type_short_name(
self,
gt_type_short_name: str,
config_factory: TipConfigDefaultParamsFactory,
serialize_for_cli: bool = False,
) -> List[TableRow] | List[ObjectifiedElement]:
# check and see if there is already a record for the provided gt type in
# the tip_config table
gt_config = await self.get_by_gt_type(gt_type_short_name)
# if records already exist, print a warning and skip adding defaults
if len(gt_config) > 0:
msg = colored.yellow(
f"\n{gt_type_short_name} already exists in dbo.Tip_Config. Skipping adding defaults..."
)
response = []
else:
sql_parameters = config_factory.build_sql_params(gt_type_short_name)
response = await self.call_hq(
ADD_DEFAULT_TIP_CONFIG_STORED_PROC, sql_parameters
)
msg = colored.green(
f"\n{gt_type_short_name} does not exist in dbo.Tip_Config. Adding defaults..."
)
if serialize_for_cli:
textui.puts(msg)
response = self.serialize_for_cli(response)
return response
[docs]
async def add_default_records_for_gt_type_short_names(
self,
endpoint_url: str,
is_enabled: bool = IS_ENABLED_DEFAULT,
is_active_mode: bool = IS_ACTIVE_MODE_DEFAULT,
serialize_for_cli: bool = False,
) -> List[List[TableRow] | List[ObjectifiedElement]]:
# these will be the same across gt types so initialize outside
# the loop to reuse
config_factory = TipConfigDefaultParamsFactory(
endpoint_url=endpoint_url,
is_enabled=is_enabled,
is_active_mode=is_active_mode,
)
tasks = [
self.add_default_record_for_gt_type_short_name(
gt_type_short_name=gt_type_short_name,
config_factory=config_factory,
serialize_for_cli=serialize_for_cli,
)
for gt_type_short_name in GT_SHORT_NAME_TO_STORED_PROC_MAP.keys()
]
return await asyncio.gather(*tasks)