from argparse import _SubParsersAction
from functools import partial
from typing import List
from enum import Enum
from dataclasses import dataclass, fields
from lxml.objectify import IntElement, StringElement
from q2_sdk.core.dynamic_imports import (
api_ExecuteStoredProcedure as ExecuteStoredProcedure,
)
from q2_sdk.tools.decorators import dev_only
from q2_sdk.hq.models.hq_params.stored_procedure import Param
from .db_object import DbObject
from .representation_row_base import RepresentationRowBase
[docs]
class HADEDataTypes(Enum):
Boolean = "Boolean"
Currency = "Currency"
Date = "Date"
Number = "Number"
Rate = "Rate"
String = "String"
[docs]
@dataclass
class CreateHADEParams:
hade_desc: str
hade_data_type: HADEDataTypes | str
hade_name: str
host_account_click_type_id: int | None = None
click_type_attributes: str | None = None
ui_text_element: str | None = None
ui_text_element_id: int | None = None
additional_desc_ui_text_element: str | None = None
additional_desc_ui_text_element_id: int | None = None
is_editable: bool = False
[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]
class HostAccountDataElementRow(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"
[docs]
class HostAccountDataElement(DbObject):
GET_BY_NAME_KEY = "HADEName"
NAME = "HostAccountDataElement"
REPRESENTATION_ROW_CLASS = HostAccountDataElementRow
[docs]
def add_arguments(self, parser: _SubParsersAction):
subparser = parser.add_parser("get_host_account_data_elements")
subparser.set_defaults(parser="get_host_account_data_elements")
subparser.set_defaults(func=partial(self.get, serialize_for_cli=True))
[docs]
@dev_only
async def get(self, serialize_for_cli=False) -> List[HostAccountDataElementRow]:
"""Note: this only works in the dev environment"""
response = await self.call_hq(
"sdk_GetHostAccountDataElement", ExecuteStoredProcedure.SqlParameters([])
)
if serialize_for_cli:
columns = ["HADE_ID", "HADEDesc", "HADEDataType", "HADEName"]
response = self.serialize_for_cli(response, columns)
return response
[docs]
async def create(self, data_element_param: CreateHADEParams) -> None:
sql_params = data_element_param.build_stored_proc_params()
await self.call_hq(
"sdk_CreateHostAccountDataElement",
ExecuteStoredProcedure.SqlParameters(sql_params),
)