from dataclasses import dataclass, fields
from enum import Enum
from typing import Any
from q2_sdk.core.dynamic_imports import (
api_ExecuteStoredProcedure as ExecuteStoredProcedure,
)
from q2_sdk.hq.db.db_object import DbObject
from q2_sdk.hq.models.hq_params.stored_procedure import Param
from q2_sdk.hq.table_row import TableRow
[docs]
@dataclass
class HADEParams:
[docs]
def build_stored_proc_params(self):
sql_params = []
for field in fields(self):
name = field.name
value = getattr(self, name)
if isinstance(value, Enum):
value = value.value
if value is not None:
this_type = self._get_sql_data_type(value)
Param(value, this_type, name).add_to_param_list(sql_params)
return sql_params
def _get_sql_data_type(self, value):
data_type = None
match value:
case bool():
data_type = ExecuteStoredProcedure.DataType.Bit
case int():
data_type = ExecuteStoredProcedure.DataType.Int
case str():
data_type = ExecuteStoredProcedure.DataType.VarChar
return data_type
[docs]
@dataclass
class GetHADEParams(HADEParams):
HADE_ID: int | None = None
HADEDataType: str | None = None
HADEName: str | None = None
HostAccountClickTypeID: int | None = None
UiTextElementID: int | None = None
[docs]
@dataclass
class CreateHADEParams(HADEParams):
HADEDesc: str
HADEDataType: str
HADEName: str
HostAccountClickTypeID: int | None = None
ClickTypeAttributes: str | None = None
UiTextElementID: int | None = None
AdditionalDescUiTextElementID: int | None = None
IsEditable: bool | None = False
ExcludeInUserAccountResponse: bool | None = False
[docs]
@dataclass
class UpdateHADEParams(HADEParams):
HADE_ID: int | None
HADEDesc: str | None = None
HADEDataType: str | None = None
HADEName: str | None = None
HostAccountClickTypeID: int | None = None
ClickTypeAttributes: str | None = None
UiTextElementID: int | None = None
AdditionalDescUiTextElementID: int | None = None
IsEditable: bool | None = False
ExcludeInUserAccountResponse: bool | None = False
[docs]
class HostAccountDataElementWithReferencesRow(TableRow):
HADE_ID: str
HADEDesc: str
HADEDataType: str
HADEName: str
HostAccountClickTypeID: int
ClickTypeAttributes: str
UiTextElementID: int
AdditionalDescUiTextElementID: int
IsEditable: bool
ExcludeInUserAccountResponse: bool
[docs]
class HostAccountDataElementWithReferences(DbObject):
NAME = "HostAccountDataElementWithReferences"
REPRESENTATION_ROW_CLASS = HostAccountDataElementWithReferencesRow
[docs]
async def get(
self,
hade_params: GetHADEParams,
serialize_for_cli=False,
) -> list[HostAccountDataElementWithReferencesRow]:
sql_params = hade_params.build_stored_proc_params()
response = await self.call_hq(
"sdk_GetHostAccountDataElementWithReferences",
ExecuteStoredProcedure.SqlParameters(sql_params),
)
if serialize_for_cli:
fields_to_truncate = []
columns = [
"HADE_ID",
"HADEDesc",
"HADEDataType",
"HADEName",
"HostAccountClickTypeID",
"ClickTypeAttributes",
"IsEditable",
"ExcludeInUserAccountResponse",
"UiTextElementID",
"AdditionalDescUiTextElementID",
]
response = self.serialize_for_cli(
response,
fields_to_display=columns,
fields_to_truncate=fields_to_truncate,
)
return response
[docs]
async def create(self, hade_params: CreateHADEParams) -> Any:
"""
Create new HostAccountDataElement objects from the provided parameters and references
"""
sql_params = hade_params.build_stored_proc_params()
response = await self.call_hq(
"sdk_CreateHostAccountDataElementWithReferences",
ExecuteStoredProcedure.SqlParameters(sql_params),
)
return response
[docs]
async def update(self, hade_params: UpdateHADEParams) -> Any:
"""
Update HostAccountDataElement objects based on the provided parameters and references
"""
sql_params = hade_params.build_stored_proc_params()
response = await self.call_hq(
"sdk_UpdateHostAccountDataElementWithReferences",
ExecuteStoredProcedure.SqlParameters(sql_params),
)
return response