from dataclasses import dataclass, fields
from enum import Enum
from lxml.objectify import BoolElement, IntElement, StringElement
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.db.representation_row_base import RepresentationRowBase
from q2_sdk.hq.models.hq_params.stored_procedure import Param
[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(RepresentationRowBase):
HADE_ID: StringElement = "HADE_ID"
HADEDesc: StringElement = "HADEDesc"
HADEDataType: StringElement = "HADEDataType"
HADEName: StringElement = "HADEName"
HostAccountClickTypeID: IntElement = "HostAccountClickTypeID"
ClickTypeAttributes: StringElement = "ClickTypeAttributes"
UiTextElementID: IntElement = "UiTextElementID"
AdditionalDescUiTextElementID: IntElement = "AdditionalDescUiTextElementID"
IsEditable: BoolElement = "IsEditable"
ExcludeInUserAccountResponse: BoolElement = "ExcludeInUserAccountResponse"
[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