from argparse import _SubParsersAction
from functools import partial
from typing import List, Optional
from q2_sdk.core.dynamic_imports import (
api_ExecuteStoredProcedure as ExecuteStoredProcedure,
)
from q2_sdk.hq.models.hq_credentials import HqCredentials
from q2_sdk.hq.table_row import TableRow
from ..models.hq_params.stored_procedure import ParamsBuilder
from ..models.instant_payments import InstantPaymentsParams
from .db_object import DbObject
from .transaction_type import TransactionType, TransactionTypeRow
D_TYPES = ExecuteStoredProcedure.DataType
[docs]
class InstantPaymentsVendorsRow(TableRow):
VendorID: int
ShortName: str
ServiceURL: str
StringIdentifier: str
GetHostAccountIDStoredProcName: str
[docs]
class InstantPaymentsDataTypeRow(TableRow):
DataTypeID: int
ShortName: str
AdditionalValueLength: int
[docs]
class InstantPaymentsRequestTypeRow(TableRow):
RequestTypeID: int
ShortName: str
GeneratedTransactionTypeID: int
EnforceLimits: bool
BitFlagValue: int
VendorID: int
CanStartOutgoingRTPConversation: bool
ReplyParentRequestTypes: int
RequiredParentDirectionOutgoing: bool
AllowExpiryDate: bool
RequireExpiryDate: bool
AllowRequestedExecutionDate: bool
RequireRequestedExecutionDate: bool
AllowAmountModificationAllowed: bool
RequireAmountModificationAllowed: bool
ExternalTransactionTypeID: int
[docs]
class InstantPayments(DbObject):
# GET_BY_NAME_KEY = "column in the db response"
NAME = "InstantPayments"
# REPRESENTATION_ROW_CLASS = InstantPaymentsRow
def __init__(
self,
logger,
hq_credentials: Optional[HqCredentials] = None,
ret_table_obj: bool = None,
):
super().__init__(logger, hq_credentials, ret_table_obj)
self._vendor = None
self._transaction_type = None
[docs]
def add_arguments(self, parser: _SubParsersAction):
subparser = parser.add_parser("get_instant_payments_vendors")
subparser.set_defaults(parser="get_instant_payments_vendors")
subparser.set_defaults(func=partial(self.get_vendors, serialize_for_cli=True))
subparser = parser.add_parser("get_instant_payments_data_types")
subparser.set_defaults(parser="get_instant_payments_data_types")
subparser.set_defaults(
func=partial(self.get_data_types, serialize_for_cli=True)
)
subparser = parser.add_parser("get_instant_payments_request_types")
subparser.set_defaults(parser="get_instant_payments_request_types")
subparser.set_defaults(
func=partial(self.get_request_types, serialize_for_cli=True)
)
subparser = parser.add_parser("add_instant_payments_vendor")
subparser.set_defaults(parser="add_instant_payments_vendor")
subparser.add_argument("short_name", help="Q2_RTPVendor.ShortName")
subparser.add_argument("url", help="Q2_RTPVendor.ServiceURL")
subparser.set_defaults(func=partial(self.create_vendor))
subparser = parser.add_parser("add_instant_payments_data_type")
subparser.set_defaults(parser="add_instant_payments_data_type")
subparser.set_defaults(func=partial(self.create_data_type))
subparser.add_argument("short_name", help="Q2_RTPDataType.ShortName")
subparser = parser.add_parser("remove_instant_payments_vendor")
subparser.set_defaults(parser="remove_instant_payments_vendor")
subparser.set_defaults(func=partial(self.remove_vendor))
subparser.add_argument("short_name", help="Q2_RTPVendor.ShortName")
subparser = parser.add_parser("remove_instant_payments_data_type")
subparser.set_defaults(parser="remove_instant_payments_data_type")
subparser.set_defaults(func=partial(self.remove_data_type))
subparser.add_argument("short_name", help="Q2_RTPDataType.ShortName")
subparser = parser.add_parser("remove_instant_payments_request_type")
subparser.set_defaults(parser="remove_instant_payments_request_type")
subparser.set_defaults(func=partial(self.remove_request_type))
subparser.add_argument("short_name", help="Q2_RTPRequestType.ShortName")
@property
def transaction_type(self) -> TransactionType:
if not self._transaction_type:
self._transaction_type = TransactionType(
self.logger, hq_credentials=self.hq_credentials, ret_table_obj=True
)
return self._transaction_type
[docs]
async def get_vendors(
self, serialize_for_cli=False
) -> List[InstantPaymentsVendorsRow]:
response = await self.call_hq(
"sdk_GetInstantPaymentsVendors",
representation_class_override=InstantPaymentsVendorsRow,
)
if serialize_for_cli:
response = self.serialize_for_cli(response)
return response
[docs]
async def get_vendor_by_name(self, name: str) -> InstantPaymentsVendorsRow:
return await self.get_by_name(
name, get_by_name_key="ShortName", get_func=self.get_vendors
)
[docs]
async def get_data_types(
self, serialize_for_cli=False
) -> List[InstantPaymentsDataTypeRow]:
response = await self.call_hq(
"sdk_GetInstantPaymentsDataTypes",
representation_class_override=InstantPaymentsDataTypeRow,
)
if serialize_for_cli:
response = self.serialize_for_cli(response)
return response
[docs]
async def get_request_types(
self, serialize_for_cli=False
) -> List[InstantPaymentsRequestTypeRow]:
response = await self.call_hq(
"sdk_GetInstantPaymentsRequestTypes",
representation_class_override=InstantPaymentsRequestTypeRow,
)
if serialize_for_cli:
response = self.serialize_for_cli(
response,
fields_to_display=[
"RequestTypeID",
"ShortName",
"VendorID",
"GeneratedTransactionTypeID",
],
)
return response
[docs]
async def create_vendor(self, short_name: str, url: str):
return await self.call_hq(
"sdk_AddInstantPaymentsVendor",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"short_name",
short_name,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar, "service_url", url
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"stored_proc",
"rtp_GetHostAccountIDByActInternal",
),
]),
)
[docs]
async def create_data_type(self, short_name: str):
return await self.call_hq(
"sdk_AddInstantPaymentsDataType",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"short_name",
short_name,
)
]),
)
[docs]
async def create_request_type(
self,
short_name: str,
generated_transaction_type: str,
enforce_limits: bool,
bitflag_value: int,
vendor_name: str,
can_start_outgoing_rtp_conversation: bool,
require_parent_direction_outgoing: bool = False,
reply_parent_request_types: Optional[int] = None,
external_transaction_type: Optional[int] = None,
):
transaction_type: TransactionTypeRow = await self.transaction_type.get_by_name(
generated_transaction_type
)
transaction_type_id: int = transaction_type.TransactionTypeID
vendor = await self.get_vendor_by_name(vendor_name)
vendor_id: int = vendor.VendorID
params_builder = (
ParamsBuilder()
.add_param(D_TYPES.VarChar, "short_name", short_name)
.add_param(
D_TYPES.Int, "generated_transaction_type_id", int(transaction_type_id)
)
.add_param(D_TYPES.Bit, "enforce_limits", enforce_limits)
.add_param(D_TYPES.Int, "bit_flag_value", int(bitflag_value))
.add_param(D_TYPES.Int, "vendor_id", int(vendor_id))
.add_param(
D_TYPES.Bit,
"can_start_outgoing_rtp_conversation",
can_start_outgoing_rtp_conversation,
)
.add_param(
D_TYPES.Bit,
"required_parent_direction_outgoing",
require_parent_direction_outgoing,
)
)
if reply_parent_request_types:
params_builder.add_param(
D_TYPES.Int,
"reply_parent_request_types",
int(reply_parent_request_types),
)
if external_transaction_type:
params_builder.add_param(
D_TYPES.Int,
"external_transaction_type_id",
int(external_transaction_type),
)
return await self.call_hq(
"sdk_AddInstantPaymentsRequestType",
ExecuteStoredProcedure.SqlParameters(params_builder.build()),
)
[docs]
async def update_request_type(
self,
short_name: str,
vendor_name: str,
instant_payment_params: InstantPaymentsParams,
update_vendor=True,
):
vendor = await self.get_vendor_by_name(vendor_name)
vendor_id = vendor.VendorID
params_builder = (
ParamsBuilder()
.add_param(D_TYPES.VarChar, "short_name", short_name)
.add_param(D_TYPES.Int, "vendor_id", vendor_id)
.add_param(
D_TYPES.Bit, "enforce_limits", instant_payment_params.enforce_limits
)
.add_param(
D_TYPES.Bit, "allow_expiry_date", instant_payment_params.allow_expiry
)
.add_param(
D_TYPES.Bit,
"require_expiry_date",
instant_payment_params.require_expiry,
)
.add_param(
D_TYPES.Bit,
"allow_requested_execution_date",
instant_payment_params.allow_requested_execution,
)
.add_param(
D_TYPES.Bit,
"allow_amount_modification_allowed",
instant_payment_params.allow_amount_modification,
)
.add_param(
D_TYPES.Bit,
"require_amount_modification_allowed",
instant_payment_params.require_amount_modification,
)
.add_param(D_TYPES.Bit, "update_vendor", update_vendor)
)
return await self.call_hq(
"sdk_UpdateInstantPaymentsRequestType",
ExecuteStoredProcedure.SqlParameters(params_builder.build()),
)
[docs]
async def remove_vendor(self, short_name: str):
return await self.call_hq(
"sdk_RemoveInstantPaymentsVendor",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"short_name",
short_name,
)
]),
)
[docs]
async def remove_data_type(self, short_name: str):
return await self.call_hq(
"sdk_RemoveInstantPaymentsDataType",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"short_name",
short_name,
)
]),
)
[docs]
async def remove_request_type(self, short_name: str):
return await self.call_hq(
"sdk_RemoveInstantPaymentsRequestType",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"short_name",
short_name,
)
]),
)