Source code for q2_sdk.hq.db.notification_status

from argparse import _SubParsersAction
from functools import partial
from typing import List

from q2_sdk.core.dynamic_imports import (
    api_ExecuteStoredProcedure as ExecuteStoredProcedure,
)
from q2_sdk.hq.models.hq_params.stored_procedure import Param
from q2_sdk.hq.table_row import TableRow

from .db_object import DbObject

D_TYPES = ExecuteStoredProcedure.DataType


[docs] class NotificationStatusRow(TableRow): NotificationStatusID: str ShortName: str DisplayName: str
[docs] class NotificationStatus(DbObject): GET_BY_NAME_KEY = "ShortName" NAME = "NotificationStatus" REPRESENTATION_ROW_CLASS = NotificationStatusRow
[docs] def add_arguments(self, parser: _SubParsersAction): subparser = parser.add_parser("get_notification_status") subparser.set_defaults(parser="get") subparser.set_defaults(func=partial(self.get, serialize_for_cli=True)) subparser.add_argument( "-i", "--notification-status-id", type=int, help="Q2_NotificationStatus.NotificationStatusID", )
[docs] async def get( self, notification_status_id: int = None, serialize_for_cli=False ) -> List[NotificationStatusRow]: """ Retrieves notification status rows. :param notification_status_id: The notification status id. If not provided, all rows will be returned. """ sql_params = [] if notification_status_id: notification_status_id = int(notification_status_id) Param( notification_status_id, D_TYPES.Int, "notification_status_id" ).add_to_param_list(sql_params) response = await self.call_hq( "sdk_GetNotificationStatus", ExecuteStoredProcedure.SqlParameters(sql_params), use_json=False, ) if serialize_for_cli: response = self.serialize_for_cli(response) return response