from typing import List
from q2_sdk.core.dynamic_imports import (
api_ExecuteStoredProcedure as ExecuteStoredProcedure,
)
from q2_sdk.hq.table_row import TableRow
from ..models.hq_params.stored_procedure import ParamsBuilder
from .db_object import DEFAULT, DbObject
D_TYPES = ExecuteStoredProcedure.DataType
[docs]
class ObjectGroupDataRow(TableRow):
ObjectGroupID: str
ObjectGroupTypeID: int
ShortName: str
Data: str
Alias: str
UserID: int
ObjectGroupTypeID1: int
ShortName1: str
UiTextElementID1: int
[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 = (
ParamsBuilder()
.add_param(D_TYPES.VarChar, "ShortName", short_name)
.add_param(D_TYPES.VarChar, "ObjectGroupType", object_group_type)
.add_param(D_TYPES.VarChar, "Alias", alias)
.add_param(D_TYPES.VarChar, "Data", data)
.add_param(D_TYPES.VarChar, "Description", description)
.add_param(D_TYPES.Int, "UserId", user_id)
.add_param(D_TYPES.VarChar, "UiTextElement", ui_text_element)
.add_param(D_TYPES.Int, "CustomerId", customer_id)
.build()
)
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 = (
ParamsBuilder()
.add_param(D_TYPES.Int, "ObjectGroupId", object_group_id)
.add_param(
D_TYPES.VarChar,
"ShortName",
str(object_row[0].ShortName)
if short_name == DEFAULT
else str(short_name),
)
.add_param(
D_TYPES.Int,
"ObjectGroupTypeId",
int(object_row[0].ObjectGroupTypeID)
if object_group_type_id == DEFAULT
else object_group_type_id,
)
.add_param(
D_TYPES.VarChar,
"Alias",
str(object_row[0].Alias) if alias == DEFAULT else alias,
)
.add_param(
D_TYPES.VarChar,
"Data",
str(object_row[0].Data) if data == DEFAULT else str(data),
)
.add_param(
D_TYPES.Int,
"UserId",
int(object_row[0].UserID) if user_id == DEFAULT else user_id,
)
.build()
)
await self.call_hq(
"sdk_UpdateObjectGroupData", ExecuteStoredProcedure.SqlParameters(params)
)