from __future__ import annotations
from argparse import _SubParsersAction
from dataclasses import dataclass
from functools import partial
from typing import List, Optional
from lxml.objectify import IntElement, StringElement
from q2_sdk.hq.models.hq_params.stored_procedure import Param
from q2_sdk.hq.db.group_to_zone import GroupToZone
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
SqlParameters = ExecuteStoredProcedure.SqlParameters
SqlParam = ExecuteStoredProcedure.SqlParam
DataType = ExecuteStoredProcedure.DataType
[docs]
@dataclass
class ZoneContext:
"""If there is a zone context at all, even with null zone_id, the response will include zone data"""
zone_id: Optional[int] = None
[docs]
@staticmethod
def from_zone_id(zone_id: int) -> ZoneContext:
return ZoneContext(zone_id)
[docs]
@staticmethod
async def from_group_id(group_id: int) -> ZoneContext:
db_obj = GroupToZone(None, ret_table_obj=True)
zones = await db_obj.get()
for row in zones:
if row.GroupID == group_id:
return ZoneContext(row.ZoneID)
raise ZoneNotFoundError
[docs]
class SystemPropertyDataRow(RepresentationRowBase):
"""Data from the Q2_SystemPropertyData table"""
SystemPropertyDataID: IntElement = "SystemPropertyDataID"
PropertyID: IntElement = "PropertyID"
PropertyValue: StringElement = "PropertyValue"
PropertyName: StringElement = "PropertyName"
PropertyLongName: StringElement = "PropertyLongName"
PropertyDataType: StringElement = "PropertyDataType"
Weight: IntElement = "PropertyID"
UISourceID: IntElement = "UiSourceID"
ProductTypeID: IntElement = "ProductTypeID"
ProductID: IntElement = "ProductID"
GroupID: IntElement = "GroupID"
HADE_ID: IntElement = "HADE_ID"
[docs]
class ZonedSystemPropertyDataRow(RepresentationRowBase):
"""Data from the Q2_ZoneSystemPropertyData table"""
ZoneSystemPropertyDataID: IntElement = "ZoneSystemPropertyDataID"
SystemPropertyDataID: IntElement = "SystemPropertyDataID"
ZoneId: IntElement = "ZoneId"
PropertyID: IntElement = "PropertyID"
PropertyValue: StringElement = "PropertyValue"
PropertyName: StringElement = "PropertyName"
PropertyLongName: StringElement = "PropertyLongName"
PropertyDataType: StringElement = "PropertyDataType"
Weight: IntElement = "PropertyID"
UISourceID: IntElement = "UiSourceID"
ProductTypeID: IntElement = "ProductTypeID"
ProductID: IntElement = "ProductID"
GroupID: IntElement = "GroupID"
HADE_ID: IntElement = "HADE_ID"
[docs]
class SystemPropertyData(DbObject):
GET_BY_NAME_KEY = "PropertyName"
NAME = "SystemPropertyData"
REPRESENTATION_ROW_CLASS = SystemPropertyDataRow
[docs]
def add_arguments(self, parser: _SubParsersAction):
subparser = parser.add_parser("get_system_property_data")
subparser.set_defaults(parser="get")
subparser.set_defaults(func=partial(self.get, serialize_for_cli=True))
subparser.add_argument(
"name", nargs="?", help="Q2_SystemPropertyDataElements.PropertyName prefix"
)
subparser.add_argument("-g", "--group-id", help="Q2_UserPropertyData.GroupID")
subparser.add_argument(
"--strict",
action="store_true",
help="Only show exact matches for group/product/etc",
)
subparser.add_argument(
"--return-count",
type=int,
help="Max number of matches to return",
default=10,
)
subparser.add_argument("--hade-id", help="Q2_SystemPropertyData.HADE_ID")
subparser.add_argument(
"--no-trunc", action="store_true", help="Do not truncate PropertyLongName"
)
group = subparser.add_mutually_exclusive_group()
group.add_argument(
"--zone", dest="zone_by_id", type=int, help="Filter by zone_id"
)
group.add_argument(
"--zone-by-group",
dest="zone_by_group",
type=int,
help="Filter by zone_id, but look it up by group",
)
group.add_argument(
"--include-zones",
dest="include_zones",
action="store_true",
help="Include zoned overrides",
)
subparser = parser.add_parser("get_system_property_data_by_id")
subparser.set_defaults(parser="get_by_id")
subparser.set_defaults(func=partial(self.get_by_id, serialize_for_cli=True))
subparser.add_argument(
"property_id", nargs="?", help="Q2_SystemPropertyDataElements.PropertyID"
)
subparser = parser.add_parser("add_system_property_data")
subparser.set_defaults(parser="add")
subparser.set_defaults(func=partial(self.add, serialize_for_cli=True))
subparser.add_argument(
"property_name", help="Q2_SystemPropertyDataElements.PropertyName"
)
subparser.add_argument(
"property_value", help="Q2_SystemPropertyData.PropertyValue"
)
subparser.add_argument(
"--product_type_name", help="Q2_ProductType.ProductTypeName"
)
subparser = parser.add_parser("update_system_property_data")
subparser.set_defaults(parser="update")
subparser.set_defaults(func=partial(self.update))
subparser.add_argument(
"property_name", help="Q2_SystemPropertyDataElements.PropertyName"
)
subparser.add_argument(
"property_value", help="Q2_SystemPropertyData.PropertyValue"
)
subparser.add_argument("--group_id", help="Q2_Group.GroupID")
def _get_system_property_params(
self,
name: Optional[str],
group_id: int,
hade_id: Optional[int],
return_count=10,
strict=False,
) -> SqlParameters:
return SqlParameters([
SqlParam(DataType.VarChar, "PropertyName", name),
SqlParam(DataType.Int, "GroupID", group_id),
SqlParam(DataType.Int, "HADE_ID", hade_id),
SqlParam(DataType.Int, "ReturnCount", return_count),
SqlParam(DataType.Bit, "Strict", str(strict)),
])
[docs]
async def get(
self,
name: Optional[str] = None,
group_id: Optional[int] = None,
hade_id: Optional[int] = None,
return_count=10,
no_trunc=False,
strict=False,
zone_context: Optional[ZoneContext | int] = None,
serialize_for_cli=False,
**kwargs,
) -> List[SystemPropertyDataRow]:
if name is None:
name = ""
truncate = not no_trunc
sql_params: SqlParameters = self._get_system_property_params(
name, group_id, hade_id, return_count, strict
)
response = await self.call_hq(
"sdk_GetSystemPropertyData", sql_parameters=sql_params
)
if kwargs.get("zone_by_group"):
zone_context = await ZoneContext.from_group_id(kwargs["zone_by_group"])
elif kwargs.get("zone_by_id"):
zone_context = ZoneContext.from_zone_id(kwargs["zone_by_id"])
elif kwargs.get("include_zones"):
zone_context = ZoneContext()
if zone_context:
if zone_context.zone_id:
sql_params.sql_param.append(
SqlParam(DataType.Int, "ZoneId", zone_context.zone_id)
)
zone_response = await self.call_hq(
"sdk_GetZoneSystemPropertyData",
sql_parameters=sql_params,
representation_class_override=ZonedSystemPropertyDataRow,
)
# Overwrite non-zone rows if a specific zone has been requested
if zone_context.zone_id:
for z_row in zone_response:
for row in response:
if row.PropertyID == z_row.PropertyID:
response.remove(row)
response.extend(zone_response)
if serialize_for_cli:
fields_to_truncate = []
if truncate:
fields_to_truncate = ["PropertyValue"]
columns = [
"SystemPropertyDataID",
"PropertyID",
"PropertyName",
"PropertyDataType",
"PropertyValue",
]
if zone_context:
columns.remove("SystemPropertyDataID")
columns.append("ZoneID")
response = self.serialize_for_cli(
response,
fields_to_display=columns,
fields_to_truncate=fields_to_truncate,
)
return response
[docs]
async def get_by_id(self, property_id: int, serialize_for_cli=False):
if not isinstance(property_id, int):
raise ValueError("Please provide a valid property_id")
response = await self.call_hq(
"sdk_GetSystemPropertyDataById",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int, "PropertyID", property_id
)
]),
)
if serialize_for_cli:
fields_to_truncate = ["PropertyValue"]
columns = [
"SystemPropertyDataID",
"PropertyID",
"PropertyName",
"PropertyDataType",
"PropertyValue",
]
response = self.serialize_for_cli(
response,
fields_to_display=columns,
fields_to_truncate=fields_to_truncate,
)
return response
[docs]
@dev_only
async def update(
self,
property_name: str,
property_value: str | bool,
group_id: Optional[int] = None,
):
list_of_params = [
Param(property_name, DataType.VarChar, "PropertyName"),
Param(str(property_value), DataType.VarChar, "PropertyValue"),
]
if group_id:
list_of_params.append(Param(group_id, DataType.Int, "GroupID"))
sql_params = []
for item in list_of_params:
item.add_to_param_list(sql_params)
return await self.call_hq(
"sdk_UpdateSystemPropertyData",
ExecuteStoredProcedure.SqlParameters(sql_params),
)
[docs]
@dev_only
async def add(
self,
property_name: str,
property_value: str,
product_type_name: str | None = None,
serialize_for_cli=False,
):
await self.call_hq(
"sdk_AddSystemPropertyData",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"propertyName",
property_name,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"propertyValue",
property_value,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"productTypeName",
product_type_name,
),
]),
)
[docs]
class ZoneNotFoundError(Exception):
"""No zone found"""