from argparse import _SubParsersAction
from functools import partial
from typing import List, 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 AlertDefinitionRow(TableRow):
AlertDefinitionID: int
AlertTypeID: int
UserID: int
Enabled: bool
SendSecureMessage: bool
OneTimeAlert: bool
NotificationTypeID: int
NotificationTime: str
LastAlerted: str
CreateDate: str
[docs]
class AlertDefinition(DbObject):
NAME = "AlertDefinition"
REPRESENTATION_ROW_CLASS = AlertDefinitionRow
[docs]
def add_arguments(self, parser: _SubParsersAction):
sub = parser.add_parser("get_alert_definition")
sub.add_argument(
"alert_type_short_name",
type=str,
help="Q2_AlertType.ShortName",
)
sub.set_defaults(parser="get_alert_definition")
sub.set_defaults(func=partial(self.get, serialize_for_cli=True))
sub = parser.add_parser("disable_alert_definition_by_type")
sub.add_argument(
"alert_type_short_name",
type=str,
help="Q2_AlertType.ShortName",
)
sub.set_defaults(parser="disable_alert_definition_by_type")
sub.set_defaults(func=partial(self.disable_by_alert_type))
[docs]
async def get(
self,
alert_type_short_name: str,
serialize_for_cli: bool = False,
) -> Union[List[AlertDefinitionRow], str]:
if (
not isinstance(alert_type_short_name, str)
or not alert_type_short_name.strip()
):
raise ValueError("alert_type_short_name must be a non-empty string")
response = await self.call_hq(
"sdk_GetAlertDefinition",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"alert_type_short_name",
alert_type_short_name,
)
]),
)
if serialize_for_cli:
response = self.serialize_for_cli(
response,
[
"AlertDefinitionID",
"AlertTypeID",
"UserID",
"Enabled",
"NotificationTypeID",
],
)
return response
[docs]
async def disable_by_alert_type(self, alert_type_short_name: str):
if (
not isinstance(alert_type_short_name, str)
or not alert_type_short_name.strip()
):
raise ValueError("alert_type_short_name must be a non-empty string")
return await self.call_hq(
"sdk_DisableAlertDefinitionByAlertType",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"alert_type_short_name",
alert_type_short_name,
)
]),
)