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.tools.decorators import dev_only
from .db_object import DbObject
from .representation_row_base import RepresentationRowBase
[docs]
class AdapterInterfaceRow(RepresentationRowBase):
InterfaceID: IntElement = "InterfaceID"
ProductTypeID: IntElement = "ProductTypeID"
RequestTypeID: IntElement = "RequestTypeID"
AddressID: IntElement = "AddressID"
FailoverInterfaceID: IntElement = "FailoverInterfaceID"
Description: StringElement = "Description"
BaseTimeoutSeconds: IntElement = "BaseTimeoutSeconds"
PerRecordTimeoutSeconds: IntElement = "PerRecordTimeoutSeconds"
IsEnabled: BoolElement = "IsEnabled"
AdditionalConfigVendorID: IntElement = "AdditionalConfigVendorID"
[docs]
class AdapterInterface(DbObject):
NAME = "AdapterInterface"
GET_BY_NAME_KEY = "Description"
REPRESENTATION_ROW_CLASS = AdapterInterfaceRow
[docs]
def add_arguments(self, parser: _SubParsersAction):
subparser = parser.add_parser("get_adapter_interfaces")
subparser.set_defaults(parser="get")
subparser.set_defaults(func=partial(self.get, serialize_for_cli=True))
subparser = parser.add_parser("update_adapter_interface")
subparser.set_defaults(parser="update")
subparser.set_defaults(func=partial(self.update, serialize_for_cli=True))
subparser.add_argument(
"short_name",
help="Short name of request type from Q2_AdapterRequestTypes",
)
subparser.add_argument(
"address_id", help="Adapter address ID to link to interface"
)
subparser.add_argument(
"product_type_id", help="Value of Q2_AdapterInterfaces.ProductTypeID"
)
subparser.add_argument(
"-e", "--enabled", help="Enable adapter interface", default=True
)
subparser = parser.add_parser("add_adapter_interface")
subparser.set_defaults(parser="add")
subparser.set_defaults(func=partial(self.add, serialize_for_cli=True))
subparser.add_argument(
"request_type_short_name", help="Q2_AdapterRequestType.ShortName"
)
subparser.add_argument("address_id", help="Q2_AdapterAddresses.AddressID")
subparser.add_argument("description", help="Q2_AdapterInterface.Description")
subparser.add_argument(
"--product_type_name", help="Q2_ProductType.ProductTypeName"
)
subparser.add_argument(
"--additional_config_vendor_name", help="Q2_Vendors.VendorName"
)
[docs]
async def get(self, serialize_for_cli=False) -> List[AdapterInterfaceRow]:
response = await self.call_hq("sdk_GetAdapterInterfaces")
if serialize_for_cli:
columns = [
"InterfaceID",
"ProductTypeID",
"RequestTypeID",
"AddressID",
"FailoverInterfaceID",
"Description",
"BaseTimeoutSeconds",
"PerRecordTimeoutSeconds",
"IsEnabled",
"AdditionalConfigVendorID",
]
response = self.serialize_for_cli(response, columns)
return response
[docs]
@dev_only
async def get_by_request_type_name(self, request_type_name: str):
response = await self.call_hq(
"sdk_GetAdapterInterfacesByRequestTypeName",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
data_type=ExecuteStoredProcedure.DataType.VarChar,
name="requestTypeShortName",
value=request_type_name,
)
]),
)
return response
[docs]
async def update(
self,
short_name,
address_id,
product_type_id,
enabled=1,
vendor_id: Optional[int] = None,
serialize_for_cli=False,
):
list_of_params = [
ExecuteStoredProcedure.SqlParam(
data_type=ExecuteStoredProcedure.DataType.VarChar,
name="short_name",
value=short_name,
),
ExecuteStoredProcedure.SqlParam(
data_type=ExecuteStoredProcedure.DataType.Int,
name="address_id",
value=address_id,
),
ExecuteStoredProcedure.SqlParam(
data_type=ExecuteStoredProcedure.DataType.Bit,
name="enabled",
value=bool(int(enabled)),
),
ExecuteStoredProcedure.SqlParam(
data_type=ExecuteStoredProcedure.DataType.Int,
name="vendor_id",
value=vendor_id,
),
]
if product_type_id:
list_of_params.append(
ExecuteStoredProcedure.SqlParam(
data_type=ExecuteStoredProcedure.DataType.Int,
name="product_type_id",
value=int(product_type_id),
)
)
parameters = ExecuteStoredProcedure.SqlParameters(list_of_params)
response = await self.call_hq("sdk_UpdateAdapterInterface", parameters)
if serialize_for_cli:
self.serialize_for_cli(
response,
[
"InterfaceID",
"ProductTypeID",
"RequestTypeID",
"AddressID",
"Description",
"IsEnabled",
"AdditionalConfigVendorID",
],
)
[docs]
async def add(
self,
request_type_short_name: str,
address_id: int,
description: str,
product_type_name: Optional[str] = None,
additional_config_vendor_name: Optional[str] = None,
serialize_for_cli=False,
):
await self.call_hq(
"sdk_AddAdapterInterface",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"requestTypeShortName",
request_type_short_name,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int, "addressId", address_id
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"description",
description,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"productTypeName",
product_type_name,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"additionalConfigVendorName",
additional_config_vendor_name,
),
]),
)