Source code for q2_sdk.hq.db.alert_type

from argparse import _SubParsersAction
from functools import partial
from typing import List, Optional, Union, Sequence

from q2_sdk.core.dynamic_imports import (
    api_ExecuteStoredProcedure as ExecuteStoredProcedure,
)
from q2_sdk.hq.table_row import TableRow
from .db_object import DbObject




class AlertTypeDataRow(TableRow):
    AlertTypeID: int
    ShortName: str
    DisplayName: str
    DisplayInOnline: int





class AlertType(DbObject):
    GET_BY_NAME_KEY = "ShortName"
    NAME = "AlertType"
    REPRESENTATION_ROW_CLASS = AlertTypeDataRow



    def add_arguments(self, parser: _SubParsersAction):
        # get_alert_type
        sub = parser.add_parser("get_alert_type")
        sub.add_argument(
            "short_name", type=str, nargs="?", help="Q2_AlertType.ShortName (optional)"
        )
        sub.set_defaults(parser="get_alert_type")
        sub.set_defaults(func=partial(self.get, serialize_for_cli=True))

        # update_alert_type
        sub = parser.add_parser("update_alert_type")
        sub.add_argument("short_name", type=str, help="Q2_AlertType.ShortName")
        sub.add_argument(
            "display_in_online",
            type=int,
            choices=[0, 1],
            help="Q2_AlertType.DisplayInOnline (0 or 1)",
        )
        sub.set_defaults(parser="update_alert_type")
        sub.set_defaults(func=partial(self.update))




    async def get(
        self,
        short_name: Optional[str] = None,
        serialize_for_cli: bool = False,
    ) -> Union[List[AlertTypeDataRow], str]:
        # Build params only when short_name is provided
        sql_params = []
        if short_name:
            sql_params.append(
                ExecuteStoredProcedure.SqlParam(
                    ExecuteStoredProcedure.DataType.VarChar,
                    "short_name",
                    short_name,
                )
            )

        response: Sequence[AlertTypeDataRow] = await self.call_hq(
            "sdk_GetAlertType",
            ExecuteStoredProcedure.SqlParameters(sql_params),
        )

        if serialize_for_cli:
            response = self.serialize_for_cli(
                response,
                [
                    "AlertTypeID",
                    "ShortName",
                    "DisplayName",
                    "DisplayInOnline",
                ],
            )

        return response




    async def update(
        self,
        short_name: str,
        display_in_online: int,
    ):
        # Validate short_name
        if not short_name or not str(short_name).strip():
            raise ValueError("short_name must be provided")

        # Normalize and validate display_in_online
        try:
            parsed = int(display_in_online)
        except (TypeError, ValueError):
            raise ValueError(
                f"display_in_online must be an integer (got {display_in_online!r})"
            )

        if parsed not in (0, 1):
            raise ValueError(f"DisplayInOnline must be 0 or 1. Got {parsed}.")

        # Call the stored procedure with correct types
        result = await self.call_hq(
            "sdk_UpdateAlertType",
            ExecuteStoredProcedure.SqlParameters([
                ExecuteStoredProcedure.SqlParam(
                    ExecuteStoredProcedure.DataType.VarChar,
                    "short_name",
                    short_name,
                ),
                ExecuteStoredProcedure.SqlParam(
                    ExecuteStoredProcedure.DataType.Bit,
                    "display_in_online",
                    parsed,
                ),
            ]),
        )
        return result