from argparse import _SubParsersAction
from functools import partial
from typing import List, Optional, Union
from q2_sdk.core.dynamic_imports import (
api_ExecuteStoredProcedure as ExecuteStoredProcedure,
)
from q2_sdk.hq.table_row import TableRow
from .db_object import DbObject
[docs]
class SecAlertProfileDataRow(TableRow):
SecAlertProfileDataID: int
SecAlertProfileID: int
SecAlertProfileDescription: str
SecAlertEventID: int
ShortName: str
Enable: bool
ForceDefaultValue: bool
DefaultValue: bool
[docs]
class SecAlertProfileData(DbObject):
NAME = "SecAlertProfileData"
REPRESENTATION_ROW_CLASS = SecAlertProfileDataRow
[docs]
def add_arguments(self, parser: _SubParsersAction):
sub = parser.add_parser("get_sec_alert_profile_data")
sub.set_defaults(parser="get_sec_alert_profile_data")
sub.set_defaults(func=partial(self.get, serialize_for_cli=True))
sub.add_argument(
"--sec-alert-profile-description",
dest="sec_alert_profile_description",
help="Q2_SecAlertProfile.Description",
)
sub.add_argument(
"--sec-alert-event-short-name",
dest="sec_alert_event_short_name",
help="Q2_SecAlertEvent.ShortName",
)
sub = parser.add_parser("create_sec_alert_profile_data")
sub.set_defaults(parser="create_sec_alert_profile_data")
sub.set_defaults(func=partial(self.create))
sub.add_argument(
"--sec-alert-profile-description",
dest="sec_alert_profile_description",
help="Q2_SecAlertProfile.Description",
)
sub.add_argument(
"--sec-alert-event-short-name",
dest="sec_alert_event_short_name",
help="Q2_SecAlertEvent.ShortName",
)
sub.add_argument(
"enable",
type=int,
choices=[0, 1],
help="Q2_SecAlertProfileData.Enable (0 or 1)",
)
sub.add_argument(
"default_value",
type=int,
choices=[0, 1],
help="Q2_SecAlertProfileData.DefaultValue (0 or 1)",
)
sub.add_argument(
"force_default_value",
type=int,
choices=[0, 1],
help="Q2_SecAlertProfileData.ForceDefaultValue (0 or 1)",
)
[docs]
async def get(
self,
sec_alert_profile_description: Optional[str] = None,
sec_alert_event_short_name: Optional[str] = None,
serialize_for_cli: bool = False,
) -> Union[List[SecAlertProfileDataRow], str]:
sql_params = []
if sec_alert_profile_description is not None:
sql_params.append(
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"sec_alert_profile_description",
sec_alert_profile_description,
)
)
if sec_alert_event_short_name is not None:
sql_params.append(
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"sec_alert_event_short_name",
sec_alert_event_short_name,
)
)
response = await self.call_hq(
"sdk_GetSecAlertProfileData",
ExecuteStoredProcedure.SqlParameters(sql_params),
)
if serialize_for_cli:
response = self.serialize_for_cli(
response,
[
"SecAlertProfileDataID",
"SecAlertProfileDescription",
"ShortName",
"Enable",
"DefaultValue",
"ForceDefaultValue",
],
)
return response
[docs]
async def create(
self,
enable: int,
default_value: int,
force_default_value: int,
sec_alert_profile_description: Optional[str] = None,
sec_alert_event_short_name: Optional[str] = None,
):
if (
not isinstance(sec_alert_profile_description, str)
or not sec_alert_profile_description.strip()
):
raise ValueError("sec_alert_profile_description must be a non-empty string")
if (
not isinstance(sec_alert_event_short_name, str)
or not sec_alert_event_short_name.strip()
):
raise ValueError("sec_alert_event_short_name must be a non-empty string")
enable = self._normalize_bit(enable, "enable")
default_value = self._normalize_bit(default_value, "default_value")
force_default_value = self._normalize_bit(
force_default_value, "force_default_value"
)
return await self.call_hq(
"sdk_AddSecAlertProfileData",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"sec_alert_profile_description",
sec_alert_profile_description,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"sec_alert_event_short_name",
sec_alert_event_short_name,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Bit,
"enable",
enable,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Bit,
"default_value",
default_value,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Bit,
"force_default_value",
force_default_value,
),
]),
)
@staticmethod
def _normalize_bit(value: int, field_name: str) -> int:
try:
parsed = int(value)
except (TypeError, ValueError):
raise ValueError(f"{field_name} must be an integer with value 0 or 1")
if parsed not in (0, 1):
raise ValueError(f"{field_name} must be 0 or 1")
return parsed