import json
from argparse import _SubParsersAction
from dataclasses import dataclass, asdict
from functools import partial
from typing import Optional
from lxml.objectify import IntElement, StringElement
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 ..models.hq_params.stored_procedure import Param
from ...core.exceptions import DatabaseMissingDataError
D_TYPES = ExecuteStoredProcedure.DataType
[docs]
@dataclass
class VendorPropertyDataParams:
property_name: str
vendor_id: int
property_value: Optional[str] = None
zone_id: Optional[int] = None
group_id: Optional[int] = None
customer_id: Optional[int] = None
user_id: Optional[int] = None
user_role_id: Optional[int] = None
product_type_id: Optional[int] = None
product_id: Optional[int] = None
host_account_id: Optional[int] = None
[docs]
class VendorPropertyDataRow(TableRow):
# object name: type hinting = "column name in the db response"
VendorPropertyID: IntElement = "VendorPropertyID"
VendorPropertyElementID: IntElement = "VendorPropertyElementID"
PropertyValue: StringElement = "PropertyValue"
VendorID: IntElement = "VendorID"
VendorPropertyName: StringElement = "VendorPropertyName"
VendorName: StringElement = "VendorName"
DataType: StringElement = "DataType"
ZoneID: IntElement = "ZoneID"
GroupID: IntElement = "GroupID"
CustomerID: IntElement = "CustomerID"
UserID: IntElement = "UserID"
UserRoleID: IntElement = "UserRoleID"
ProductTypeID: IntElement = "ProductTypeID"
ProductID: IntElement = "ProductID"
HostAccountID: IntElement = "HostAccountID"
Weight: IntElement = "Weight"
[docs]
class VendorPropertyHostAccountDataRow(TableRow):
# object name: type hinting = "column name in the db response"
HostAccountID: IntElement = "HostAccountID"
PropertyValue: StringElement = "PropertyValue"
[docs]
async def common_params(vendor_property_data_params, params):
Param(vendor_property_data_params.zone_id, D_TYPES.Int, "zoneID").add_to_param_list(
params
)
Param(
vendor_property_data_params.group_id, D_TYPES.Int, "groupID"
).add_to_param_list(params)
Param(
vendor_property_data_params.customer_id, D_TYPES.Int, "customerID"
).add_to_param_list(params)
Param(vendor_property_data_params.user_id, D_TYPES.Int, "userID").add_to_param_list(
params
)
Param(
vendor_property_data_params.user_role_id, D_TYPES.Int, "userRoleID"
).add_to_param_list(params)
Param(
vendor_property_data_params.product_type_id, D_TYPES.Int, "productTypeID"
).add_to_param_list(params)
Param(
vendor_property_data_params.product_id, D_TYPES.Int, "productID"
).add_to_param_list(params)
Param(
vendor_property_data_params.host_account_id, D_TYPES.Int, "hostAccountID"
).add_to_param_list(params)
return params
[docs]
@dataclass
class HostAccountDataParams:
HostAccountID: int
ProductID: Optional[int] = None
ProductTypeID: Optional[int] = None
UserID: Optional[int] = None
UserRoleID: Optional[int] = None
CustomerID: Optional[int] = None
GroupID: Optional[int] = None
ZoneID: Optional[int] = None
[docs]
class VendorPropertyData(DbObject):
# GET_BY_NAME_KEY = "column in the db response"
NAME = "VendorPropertyData"
REPRESENTATION_ROW_CLASS = VendorPropertyDataRow
[docs]
def add_arguments(self, parser: _SubParsersAction):
subparser = parser.add_parser("get_vendor_property_data")
subparser.set_defaults(parser="get")
subparser.set_defaults(func=partial(self.get, serialize_for_cli=True))
subparser.add_argument(
"property_name", help="Q2_VendorPropertyElements.PropertyName"
)
subparser.add_argument("vendor_id", help="Q2_VendorPropertyData.VendorID_")
subparser.add_argument("-z", "--zone-id", help="Q2_Zone.ZoneID")
subparser.add_argument("-g", "--group-id", help="Q2_Group.GroupID")
subparser.add_argument("-c", "--customer-id", help="Q2_Customer.CustomerID")
subparser.add_argument("-u", "--user-id", help="Q2_User.UseID")
subparser.add_argument("-ur", "--user-role-id", help="Q2_UserRole.UserRoleID")
subparser.add_argument(
"-pt", "--product-type-id", help="Q2_ProductType.ProductTypeID"
)
subparser.add_argument("-p", "--product-id", help="Q2_Product.ProductID")
subparser.add_argument(
"-ha", "--host-account-id", help="Q2_HostAccount.HostAccountID"
)
subparser = parser.add_parser("add_vendor_property_data")
subparser.set_defaults(parser="add_vendor_property_data")
subparser.set_defaults(func=partial(self.add))
subparser.add_argument(
"property_name", help="Q2_VendorPropertyElements.PropertyName"
)
subparser.add_argument("vendor_id", help="Q2_VendorPropertyData.VendorID_")
subparser.add_argument(
"property_value", help="Q2_VendorPropertyData.PropertyValue"
)
subparser.add_argument("-z", "--zone-id", help="Q2_Zone.ZoneID")
subparser.add_argument("-g", "--group-id", help="Q2_Group.GroupID")
subparser.add_argument("-c", "--customer-id", help="Q2_Customer.CustomerID")
subparser.add_argument("-u", "--user-id", help="Q2_User.UseID")
subparser.add_argument("-ur", "--user-role-id", help="Q2_UserRole.UserRoleID")
subparser.add_argument(
"-pt", "--product-type-id", help="Q2_ProductType.ProductTypeID"
)
subparser.add_argument("-p", "--product-id", help="Q2_Product.ProductID")
subparser.add_argument(
"-ha", "--host-account-id", help="Q2_HostAccount.HostAccountID"
)
subparser = parser.add_parser("get_by_host_account_batch")
subparser.set_defaults(parser="get_by_host_account_batch")
subparser.set_defaults(
func=partial(self.get_by_host_account_batch, serialize_for_cli=True)
)
subparser.add_argument(
"property_name", help="Q2_VendorPropertyElements.PropertyName"
)
subparser.add_argument("vendor_id", help="Q2_VendorPropertyData.VendorID_")
subparser.add_argument(
"host_account_id", help="Q2_VendorPropertyData.VendorID_"
)
subparser.add_argument("-p", "--product-id", help="Q2_Product.ProductID")
subparser.add_argument(
"-pt", "--product-type-id", help="Q2_ProductType.ProductTypeID"
)
subparser.add_argument("-u", "--user-id", help="Q2_User.UseID")
subparser.add_argument("-ur", "--user-role-id", help="Q2_UserRole.UserRoleID")
subparser.add_argument("-c", "--customer-id", help="Q2_Customer.CustomerID")
subparser.add_argument("-g", "--group-id", help="Q2_Group.GroupID")
subparser.add_argument("-z", "--zone-id", help="Q2_Zone.ZoneID")
subparser = parser.add_parser("update_vendor_property_data")
subparser.set_defaults(parser="update")
subparser.set_defaults(func=partial(self.update))
subparser.add_argument(
"vendor_property_id", help="Q2_VendorPropertyData.VendorPropertyID"
)
subparser.add_argument(
"property_value", help="Q2_VendorPropertyData.PropertyValue"
)
subparser = parser.add_parser("delete_vendor_property_data")
subparser.set_defaults(parser="delete")
subparser.set_defaults(func=partial(self.delete))
subparser.add_argument(
"vendor_property_id", help="Q2_VendorPropertyData.VendorPropertyID"
)
[docs]
@staticmethod
async def common_params(vendor_property_data_params, params):
Param(
vendor_property_data_params.zone_id, D_TYPES.Int, "zoneID"
).add_to_param_list(params)
Param(
vendor_property_data_params.group_id, D_TYPES.Int, "groupID"
).add_to_param_list(params)
Param(
vendor_property_data_params.customer_id, D_TYPES.Int, "customerID"
).add_to_param_list(params)
Param(
vendor_property_data_params.user_id, D_TYPES.Int, "userID"
).add_to_param_list(params)
Param(
vendor_property_data_params.user_role_id, D_TYPES.Int, "userRoleID"
).add_to_param_list(params)
Param(
vendor_property_data_params.product_type_id, D_TYPES.Int, "productTypeID"
).add_to_param_list(params)
Param(
vendor_property_data_params.product_id, D_TYPES.Int, "productID"
).add_to_param_list(params)
Param(
vendor_property_data_params.host_account_id, D_TYPES.Int, "hostAccountID"
).add_to_param_list(params)
return params
[docs]
async def get_by_host_account_batch(
self,
property_name: str,
vendor_id: int,
host_account_data: HostAccountDataParams,
serialize_for_cli=False,
) -> [VendorPropertyHostAccountDataRow]:
input_json = json.dumps(asdict(host_account_data))
try:
response = await self.call_hq(
"Q2_VendorPropertyDataGetByHostAccount_Batch",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"propertyName",
property_name,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int, "vendorID", vendor_id
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.NVarChar,
"inputJson",
input_json,
),
]),
)
vendor_properties_json = str(response[0].findtext("VendorPropertiesJSON"))
records = json.loads(vendor_properties_json)
result = []
for rec in records:
row_elem = VendorPropertyHostAccountDataRow.from_values(**rec)
result.append(row_elem)
if serialize_for_cli:
columns = ["HostAccountID", "PropertyValue"]
result = self.serialize_for_cli(result, columns, fields_to_truncate=[])
return result
except Exception as e:
self.logger.exception(e)
[docs]
async def update(
self, vendor_property_id: int, property_value: str
) -> list[VendorPropertyDataRow]:
response = await self.call_hq(
"Q2_VendorPropertyDataUpdate",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int,
"vendorPropertyId",
vendor_property_id,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"propertyValue",
property_value,
),
]),
)
return response
[docs]
async def get(
self,
vendor_property_data_params: VendorPropertyDataParams,
serialize_for_cli=False,
) -> [VendorPropertyDataRow]:
params = []
Param(
vendor_property_data_params.property_name, D_TYPES.VarChar, "propertyName"
).add_to_param_list(params)
Param(
vendor_property_data_params.vendor_id, D_TYPES.Int, "vendorID"
).add_to_param_list(params)
await common_params(vendor_property_data_params, params)
try:
response = await self.call_hq(
"Q2_VendorPropertyDataGet", ExecuteStoredProcedure.SqlParameters(params)
)
vendor_properties_json = str(response[0].findtext("VendorPropertiesJSON"))
records = json.loads(vendor_properties_json)
result = []
for rec in records:
row_elem = VendorPropertyDataRow.from_values(**rec)
result.append(row_elem)
if serialize_for_cli:
columns = [
"VendorPropertyID",
"VendorPropertyElementID",
"ZoneID",
"GroupID",
"CustomerID",
"UserID",
"UserRoleID",
"ProductTypeID",
"ProductID",
"HostAccountID",
"PropertyValue",
"Weight",
"VendorPropertyName",
"VendorID",
"VendorName",
"DataType",
]
result = self.serialize_for_cli(result, columns, fields_to_truncate=[])
return result
except Exception as e:
self.logger.exception(e)
[docs]
async def add(self, vendor_property_data_params: VendorPropertyDataParams):
if not vendor_property_data_params.property_value:
raise DatabaseMissingDataError("Property Value is required")
params = []
Param(
vendor_property_data_params.property_name, D_TYPES.VarChar, "propertyName"
).add_to_param_list(params)
Param(
vendor_property_data_params.vendor_id, D_TYPES.Int, "vendorID"
).add_to_param_list(params)
Param(
vendor_property_data_params.property_value, D_TYPES.VarChar, "propertyValue"
).add_to_param_list(params)
await common_params(vendor_property_data_params, params)
await self.call_hq(
"Q2_VendorPropertyDataInsert", ExecuteStoredProcedure.SqlParameters(params)
)
[docs]
async def delete(self, vendor_property_id: int):
await self.call_hq(
"Q2_VendorPropertyDataDelete",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int,
"vendorPropertyId",
vendor_property_id,
)
]),
)