from typing import List
from lxml.objectify import IntElement, StringElement
from q2_sdk.core.dynamic_imports import (
api_ExecuteStoredProcedure as ExecuteStoredProcedure,
)
from .db_object import DbObject, DEFAULT
from .representation_row_base import RepresentationRowBase
from ..models.hq_params.stored_procedure import Param
D_TYPES = ExecuteStoredProcedure.DataType
[docs]
class ObjectGroupDataRow(RepresentationRowBase):
ObjectGroupID: StringElement = "ObjectGroupID"
ObjectGroupTypeID: IntElement = "ObjectGroupTypeID"
ShortName: StringElement = "ShortName"
Data: StringElement = "Data"
Alias: StringElement = "Alias"
UserID: IntElement = "UserID"
ObjectGroupTypeID1: IntElement = "ObjectGroupTypeID1"
ShortName1: StringElement = "ShortName1"
UiTextElementID1: IntElement = "UiTextElementID1"
[docs]
class ObjectGroupData(DbObject):
GET_BY_NAME_KEY = "ShortName"
NAME = "ObjectGroup"
REPRESENTATION_ROW_CLASS = ObjectGroupDataRow
[docs]
async def get(
self,
user_id: int,
object_group_type: str,
) -> List[ObjectGroupDataRow]:
response = await self.call_hq(
"sdk_GetObjectGroupData",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int, "user_id", user_id
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"object_group_type",
object_group_type,
),
]),
)
return response
[docs]
async def get_by_id(self, object_group_id: int) -> List[ObjectGroupDataRow]:
response = await self.call_hq(
"sdk_GetObjectGroupById",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int,
"object_group_id",
object_group_id,
)
]),
)
return response
[docs]
async def add(
self,
short_name: str,
object_group_type: str,
alias: str,
data: str,
description: str = None,
user_id: int = None,
ui_text_element: str = None,
customer_id=None,
) -> None:
params = []
Param(short_name, D_TYPES.VarChar, "ShortName").add_to_param_list(params)
Param(object_group_type, D_TYPES.VarChar, "ObjectGroupType").add_to_param_list(
params
)
Param(alias, D_TYPES.VarChar, "Alias").add_to_param_list(params)
Param(data, D_TYPES.VarChar, "Data").add_to_param_list(params)
Param(description, D_TYPES.VarChar, "Description").add_to_param_list(params)
Param(user_id, D_TYPES.Int, "UserId").add_to_param_list(params)
Param(ui_text_element, D_TYPES.VarChar, "UiTextElement").add_to_param_list(
params
)
Param(customer_id, D_TYPES.Int, "CustomerId").add_to_param_list(params)
response = await self.call_hq(
"sdk_AddObjectGroupData", ExecuteStoredProcedure.SqlParameters(params)
)
return response
[docs]
async def update(
self,
object_group_id: int,
short_name=DEFAULT,
object_group_type_id=DEFAULT,
alias=DEFAULT,
data=DEFAULT,
user_id=DEFAULT,
):
object_row = await ObjectGroupData(
self.logger, self.hq_credentials, ret_table_obj=None
).get_by_id(object_group_id)
params = []
Param(object_group_id, D_TYPES.Int, "ObjectGroupId").add_to_param_list(params)
Param(
str(object_row[0].ShortName) if short_name == DEFAULT else str(short_name),
D_TYPES.VarChar,
"ShortName",
).add_to_param_list(params)
Param(
int(object_row[0].ObjectGroupTypeID)
if object_group_type_id == DEFAULT
else object_group_type_id,
D_TYPES.Int,
"ObjectGroupTypeId",
).add_to_param_list(params)
Param(
str(object_row[0].Alias) if alias == DEFAULT else alias,
D_TYPES.VarChar,
"Alias",
).add_to_param_list(params)
Param(
str(object_row[0].Data) if data == DEFAULT else str(data),
D_TYPES.VarChar,
"Data",
).add_to_param_list(params)
Param(
int(object_row[0].UserID) if user_id == DEFAULT else user_id,
D_TYPES.Int,
"UserId",
).add_to_param_list(params)
await self.call_hq(
"sdk_UpdateObjectGroupData", ExecuteStoredProcedure.SqlParameters(params)
)