from __future__ import annotations
from argparse import _SubParsersAction
from datetime import datetime
from functools import partial
from typing import Any, List
from lxml.objectify import IntElement, StringElement
from q2_sdk.core.dynamic_imports import (
api_ExecuteStoredProcedure as ExecuteStoredProcedure,
)
from .db_object import DbObject
from .representation_row_base import RepresentationRowBase
DataType = ExecuteStoredProcedure.DataType
SqlParameters = ExecuteStoredProcedure.SqlParameters
SqlParam = ExecuteStoredProcedure.SqlParam
[docs]
class ConfigDataType:
Bool = "Bool"
Currency = "Currency"
Date = "Date"
File = "File"
Int = "Int"
Integer = "Integer"
Number = "Number"
String = "String"
[docs]
@staticmethod
def from_value(value: Any) -> ConfigDataType:
if isinstance(value, bool):
data_type = ConfigDataType.Bool
elif isinstance(value, int):
data_type = ConfigDataType.Integer
elif isinstance(value, float):
data_type = ConfigDataType.Number
elif isinstance(value, datetime):
data_type = ConfigDataType.Date
else:
data_type = ConfigDataType.String
return data_type
[docs]
class VendorConfigRow(RepresentationRowBase): # pragma: no cover
VendorID: IntElement = "VendorID"
ConfigID: IntElement = "ConfigID"
ConfigValue: StringElement = "ConfigValue"
ConfigName: StringElement = "ConfigName"
Description: StringElement = "Description"
DataTypeID: IntElement = "DataTypeID"
VendorName: StringElement = "VendorName"
DataType: StringElement = "DataType"
[docs]
class VendorConfig(DbObject):
NAME = "VendorConfig"
REPRESENTATION_ROW_CLASS = VendorConfigRow
[docs]
def add_arguments(self, parser: _SubParsersAction):
subparser = parser.add_parser("get_vendor_configs")
subparser.set_defaults(parser="get_vendor_configs")
subparser.set_defaults(func=partial(self.get, serialize_for_cli=True))
subparser.add_argument("vendor_name", help="Q2_Vendor.VendorName")
subparser = parser.add_parser("add_vendor_configs")
subparser.set_defaults(parser="add_vendor_configs")
subparser.set_defaults(func=partial(self.create))
subparser.add_argument("vendor_name", help="Q2_Vendor.VendorName")
subparser.add_argument(
"config_name", help="Q2_VendorConfigDataElements.ConfigName"
)
subparser.add_argument("config_value", help="Q2_VendorConfigData.ConfigValue")
subparser.add_argument("data_type", help="Q2_DataType.DataType")
subparser.add_argument(
"-d", "--description", help="Q2_VendorConfigDataElements.Description"
)
subparser = parser.add_parser("update_vendor_configs")
subparser.set_defaults(parser="update_vendor_configs")
subparser.set_defaults(func=partial(self.update))
subparser.add_argument("vendor_name", help="Q2_Vendor.VendorName")
subparser.add_argument(
"config_name", help="Q2_VendorConfigDataElements.ConfigName"
)
subparser.add_argument("config_value", help="Q2_VendorConfigData.ConfigValue")
subparser.add_argument("data_type", help="Q2_DataType.DataTypeID")
subparser.add_argument(
"-d", "--description", help="Q2_VendorConfigDataElements.Description"
)
subparser = parser.add_parser("delete_vendor_configs")
subparser.set_defaults(parser="delete_vendor_configs")
subparser.set_defaults(func=partial(self.delete))
subparser.add_argument("vendor_name", help="Q2_Vendor.VendorName")
subparser.add_argument(
"config_name", help="Q2_VendorConfigDataElements.ConfigName"
)
[docs]
async def get(
self, vendor_name: str, serialize_for_cli=False, zone_id=0
) -> List[VendorConfigRow]:
response = await self.call_hq(
"sdk_GetVendorConfig",
SqlParameters([SqlParam(DataType.VarChar, "vendor_name", vendor_name)]),
)
vendor_id = None
if response:
vendor_id = response[0].VendorID
if vendor_id and await self.hq_credentials.is_zoned():
await self._overlay_zoned_configs(response, zone_id)
if serialize_for_cli:
columns = [
"VendorID",
"VendorName",
"ConfigID",
"ConfigName",
"ConfigValue",
"DataType",
]
response = self.serialize_for_cli(response, columns)
return response
async def _overlay_zoned_configs(self, response, zone_id=0):
vendor_id = response[0].VendorID
params = [
SqlParam(DataType.Int, "ZoneID", zone_id),
SqlParam(DataType.Int, "VendorID", vendor_id.pyval),
]
zoned_response = await self.call_hq(
"Q2_ZoneVendorConfigView", sql_parameters=SqlParameters(params)
)
for zoned_item in zoned_response:
for orig_item in response:
if (
zoned_item.ConfigName == orig_item.ConfigName
and zoned_item.ConfigValue != orig_item.ConfigValue
):
self.logger.debug(
"Overriding %s with zoned variant", orig_item.ConfigName.text
)
orig_item.ConfigValue = zoned_item.ConfigValue
[docs]
async def get_all(self) -> List[VendorConfigRow]:
return await self.call_hq("sdk_GetVendorConfig")
[docs]
async def create(
self,
vendor_name: str,
config_name: str,
config_value: str,
data_type: ConfigDataType,
description: str = None,
):
parameters = [
SqlParam(DataType.VarChar, "vendor_name", vendor_name),
SqlParam(DataType.VarChar, "config_name", config_name),
SqlParam(DataType.VarChar, "config_value", config_value),
SqlParam(DataType.VarChar, "data_type", data_type),
]
if description:
parameters.append(SqlParam(DataType.VarChar, "description", description))
response = await self.call_hq("sdk_AddVendorConfig", SqlParameters(parameters))
return response
[docs]
async def update(
self,
vendor_name: str,
config_name: str,
config_value: str,
data_type: ConfigDataType,
description: str = None,
):
return await self.create(
vendor_name, config_name, config_value, data_type, description=description
)
[docs]
async def delete(self, vendor_name: str, config_name: str):
response = await self.call_hq(
"sdk_RemoveVendorConfig",
SqlParameters([
SqlParam(DataType.VarChar, "vendor_name", vendor_name),
SqlParam(DataType.VarChar, "config_name", config_name),
]),
)
return response