from argparse import _SubParsersAction
from functools import partial
from typing import List, Optional
from lxml.objectify import IntElement, StringElement, BoolElement
from q2_sdk.core.dynamic_imports import (
api_ExecuteStoredProcedure as ExecuteStoredProcedure,
)
from q2_sdk.hq.models.hq_credentials import HqCredentials
from .transaction_type import TransactionType, TransactionTypeRow
from .db_object import DbObject
from .representation_row_base import RepresentationRowBase
from ..models.hq_params.stored_procedure import Param
from ..models.instant_payments import InstantPaymentsParams
D_TYPES = ExecuteStoredProcedure.DataType
[docs]
class InstantPaymentsVendorsRow(RepresentationRowBase):
VendorID: IntElement = "VendorID"
ShortName: StringElement = "ShortName"
ServiceURL: StringElement = "ServiceURL"
StringIdentifier: StringElement = "StringIdentifier"
GetHostAccountIDStoredProcName: StringElement = "GetHostAccountIDStoredProcName"
[docs]
class InstantPaymentsDataTypeRow(RepresentationRowBase):
DataTypeID: IntElement = "DataTypeID"
ShortName: StringElement = "ShortName"
AdditionalValueLength: IntElement = "ServiceURL"
[docs]
class InstantPaymentsRequestTypeRow(RepresentationRowBase):
RequestTypeID: IntElement = "RequestTypeID"
ShortName: StringElement = "ShortName"
GeneratedTransactionTypeID: IntElement = "GeneratedTransactionTypeID"
EnforceLimits: BoolElement = "EnforceLimits"
BitFlagValue: IntElement = "BitFlagValue"
VendorID: IntElement = "VendorID"
CanStartOutgoingRTPConversation: BoolElement = "CanStartOutgoingRTPConversation"
ReplyParentRequestTypes: IntElement = "ReplyParentRequestTypes"
RequiredParentDirectionOutgoing: BoolElement = "RequiredParentDirectionOutgoing"
AllowExpiryDate: BoolElement = "AllowExpiryDate"
RequireExpiryDate: BoolElement = "RequireExpiryDate"
AllowRequestedExecutionDate: BoolElement = "AllowRequestedExecutionDate"
RequireRequestedExecutionDate: BoolElement = "RequireRequestedExecutionDate"
AllowAmountModificationAllowed: BoolElement = "AllowAmountModificationAllowed"
RequireAmountModificationAllowed: BoolElement = "RequireAmountModificationAllowed"
ExternalTransactionTypeID: IntElement = "ExternalTransactionTypeID"
[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
parameters = []
sql_params = [
Param(short_name, D_TYPES.VarChar, "short_name"),
Param(
int(transaction_type_id), D_TYPES.Int, "generated_transaction_type_id"
),
Param(enforce_limits, D_TYPES.Bit, "enforce_limits"),
Param(int(bitflag_value), D_TYPES.Int, "bit_flag_value"),
Param(int(vendor_id), D_TYPES.Int, "vendor_id"),
Param(
can_start_outgoing_rtp_conversation,
D_TYPES.Bit,
"can_start_outgoing_rtp_conversation",
),
Param(
require_parent_direction_outgoing,
D_TYPES.Bit,
"required_parent_direction_outgoing",
),
]
if reply_parent_request_types:
sql_params.append(
Param(
int(reply_parent_request_types),
D_TYPES.Int,
"reply_parent_request_types",
),
)
if external_transaction_type:
sql_params.append(
Param(
int(external_transaction_type),
D_TYPES.Int,
"external_transaction_type_id",
),
)
for item in sql_params:
item.add_to_param_list(parameters)
return await self.call_hq(
"sdk_AddInstantPaymentsRequestType",
ExecuteStoredProcedure.SqlParameters(parameters),
)
[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
parameters = []
sql_params = [
Param(short_name, D_TYPES.VarChar, "short_name"),
Param(vendor_id, D_TYPES.Int, "vendor_id"),
Param(instant_payment_params.enforce_limits, D_TYPES.Bit, "enforce_limits"),
Param(
instant_payment_params.allow_expiry, D_TYPES.Bit, "allow_expiry_date"
),
Param(
instant_payment_params.require_expiry,
D_TYPES.Bit,
"require_expiry_date",
),
Param(
instant_payment_params.allow_requested_execution,
D_TYPES.Bit,
"allow_requested_execution_date",
),
Param(
instant_payment_params.allow_amount_modification,
D_TYPES.Bit,
"allow_amount_modification_allowed",
),
Param(
instant_payment_params.require_amount_modification,
D_TYPES.Bit,
"require_amount_modification_allowed",
),
Param(update_vendor, D_TYPES.Bit, "update_vendor"),
]
for item in sql_params:
item.add_to_param_list(parameters)
return await self.call_hq(
"sdk_UpdateInstantPaymentsRequestType",
ExecuteStoredProcedure.SqlParameters(parameters),
)
[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,
)
]),
)