from typing import Any
from q2_sdk.core.dynamic_imports import (
api_ExecuteStoredProcedure as ExecuteStoredProcedure,
)
from .db_object import DbObject
from .representation_row_base import RepresentationRowBase
SqlParameters = ExecuteStoredProcedure.SqlParameters
SqlParam = ExecuteStoredProcedure.SqlParam
DataType = ExecuteStoredProcedure.DataType
[docs]
class SystemPropertyDataWithReferencesRow(RepresentationRowBase):
# Direct fields from Q2_SystemPropertyData
SystemPropertyDataID: int | None
PropertyValue: str | None
Weight: int | None
UISourceID: int | None
ProductTypeID: int | None
ProductID: int | None
GroupID: int | None
HADE_ID: int | None
PropertyID: int | None
# JSON References
Property: Any | None # JSON object: Q2_SystemPropertyDataElements
UISource: Any | None # JSON object: Q2_UISource
ProductType: Any | None # JSON object: Q2_ProductType
Product: Any | None # JSON object: Q2_Product
Group: Any | None # JSON object: Q2_Group
HADE: Any | None # JSON object: Q2_HostAccountDataElement
[docs]
class SystemPropertyDataWithReferences(DbObject):
NAME = "SystemPropertyDataWithReferences"
REPRESENTATION_ROW_CLASS = SystemPropertyDataWithReferencesRow
def _get_params_from_query(
self,
system_property_data_id: int | None = None,
property_id: int | None = None,
property_name: str | None = None,
product_id: int | None = None,
product_type_id: int | None = None,
hade_id: int | None = None,
) -> SqlParameters:
params = []
if system_property_data_id not in [None, ""]:
params.append(
SqlParam(DataType.Int, "SystemPropertyDataID", system_property_data_id)
)
if property_id not in [None, ""]:
params.append(SqlParam(DataType.Int, "PropertyID", property_id))
if property_name not in [None, ""]:
params.append(SqlParam(DataType.VarChar, "PropertyName", property_name))
if product_id not in [None, ""]:
params.append(SqlParam(DataType.Int, "ProductID", product_id))
if product_type_id not in [None, ""]:
params.append(SqlParam(DataType.Int, "ProductTypeID", product_type_id))
if hade_id not in [None, ""]:
params.append(SqlParam(DataType.Int, "HADE_ID", hade_id))
return SqlParameters(params)
[docs]
async def get(
self,
system_property_data_id: int | None = None,
property_id: int | None = None,
property_name: str | None = None,
product_id: int | None = None,
product_type_id: int | None = None,
hade_id: int | None = None,
serialize_for_cli=False,
) -> list[SystemPropertyDataWithReferencesRow]:
sql_params = self._get_params_from_query(
system_property_data_id,
property_id,
property_name,
product_id,
product_type_id,
hade_id,
)
response = await self.call_hq(
"sdk_GetSystemPropertyDataWithReferences",
sql_parameters=sql_params,
)
if serialize_for_cli:
fields_to_truncate = []
columns = [
"SystemPropertyDataID",
"UISourceID",
"PropertyID",
"PropertyName",
"PropertyDataType",
"PropertyValue",
"ProductID",
"ProductTypeID",
"GroupID",
"Hade_ID",
]
response = self.serialize_for_cli(
response,
fields_to_display=columns,
fields_to_truncate=fields_to_truncate,
)
return response
[docs]
async def add(
self,
property_name: str,
property_value: str,
ui_source_id: int | None = None,
product_type_id: int | None = None,
product_id: int | None = None,
group_id: int | None = None,
hade_id: int | None = None,
) -> Any:
"""
Insert new system property data using individual parameters.
This matches the dbo.sdk_InsertSystemPropertyDataWithReferences stored procedure.
"""
params = [
SqlParam(DataType.VarChar, "PropertyName", property_name),
SqlParam(DataType.NVarChar, "PropertyValue", property_value),
]
if ui_source_id is not None:
params.append(SqlParam(DataType.Int, "UISourceID", ui_source_id))
if product_type_id is not None:
params.append(SqlParam(DataType.Int, "ProductTypeID", product_type_id))
if product_id is not None:
params.append(SqlParam(DataType.Int, "ProductID", product_id))
if group_id is not None:
params.append(SqlParam(DataType.Int, "GroupID", group_id))
if hade_id is not None:
params.append(SqlParam(DataType.Int, "HADE_ID", hade_id))
sql_parameters = SqlParameters(params)
response = await self.call_hq(
"sdk_InsertSystemPropertyDataWithReferences",
sql_parameters=sql_parameters,
)
return response
[docs]
async def update(
self,
system_property_data_id: int,
property_name: str,
property_value: str,
ui_source_id: int | None = None,
product_type_id: int | None = None,
product_id: int | None = None,
group_id: int | None = None,
hade_id: int | None = None,
) -> Any:
"""
Update an existing system property data record using individual parameters.
This matches the dbo.sdk_UpdateSystemPropertyDataWithReferences stored procedure.
"""
params = [
SqlParam(DataType.Int, "SystemPropertyDataID", system_property_data_id),
SqlParam(DataType.VarChar, "PropertyName", property_name),
SqlParam(DataType.NVarChar, "PropertyValue", property_value),
]
if ui_source_id is not None:
params.append(SqlParam(DataType.Int, "UISourceID", ui_source_id))
if product_type_id is not None:
params.append(SqlParam(DataType.Int, "ProductTypeID", product_type_id))
if product_id is not None:
params.append(SqlParam(DataType.Int, "ProductID", product_id))
if group_id is not None:
params.append(SqlParam(DataType.Int, "GroupID", group_id))
if hade_id is not None:
params.append(SqlParam(DataType.Int, "HADE_ID", hade_id))
sql_parameters = SqlParameters(params)
response = await self.call_hq(
"sdk_UpdateSystemPropertyDataWithReferences",
sql_parameters=sql_parameters,
)
return response