from argparse import _SubParsersAction
from dataclasses import dataclass
from functools import partial
from typing import List, Optional
from lxml.etree import tostring
from lxml.objectify import E
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]
@dataclass
class NotificationAddResponse:
success: bool = False
error_message: str = ""
id: int = None
[docs]
@dataclass
class CreateNotificationParams:
NotificationTypeID: str
CreateDate: str
NotificationStatusID: str
TargetAddress: str
NotificationSubject: str
RetryCount: str = "0"
NotificationTime: Optional[str] = None
NotificationBody: Optional[str] = None
AccessCodeID: Optional[str] = None
ErrorMessage: Optional[str] = None
UserID: Optional[str] = None
TransactionID: Optional[str] = None
UiLanguageID: Optional[str] = None
HtmlBody: Optional[str] = None
TargetAddressRaw: Optional[str] = None
CountryID: Optional[str] = None
NotificationFlavorID: Optional[str] = None
ZoneID: Optional[str] = None
SessionId: Optional[str] = None
[docs]
class NotificationRow(TableRow):
NotificationID: str
NotificationTypeID: str
CreateDate: str
NotificationStatusID: str
NotificationTime: str
TargetAddress: str
NotificationSubject: str
NotificationBody: str
AccessCodeID: str
RetryCount: str
ErrorMessage: str
UserID: str
TransactionID: str
UiLanguageID: str
HtmlBody: str
TargetAddressRaw: str
CountryID: str
NotificationFlavorID: str
ZoneID: str
SessionId: str
[docs]
class Notification(DbObject):
NAME = "Notification"
REPRESENTATION_ROW_CLASS = NotificationRow
[docs]
def add_arguments(self, parser: _SubParsersAction):
subparser = parser.add_parser("get_notification")
subparser.set_defaults(parser="get")
subparser.set_defaults(func=partial(self.get, serialize_for_cli=True))
subparser.add_argument(
"-i",
"--notification-ids",
type=int,
nargs="+",
help="A space deliminated list of Q2_Notifications.NotificationID. Example: -i 1 25 13",
)
[docs]
async def get(
self, notification_ids: list[int], serialize_for_cli=False
) -> List[NotificationRow]:
"""
Retrieves notification rows based on provided notification IDs.
:param notification_ids: List of notification IDs to search.
"""
assert isinstance(notification_ids, list), (
"notification_ids param must be a list of integers"
)
sql_params = []
get_ids = [E.get(id=str(id)) for id in notification_ids]
request_xml = E.request(*get_ids)
request = tostring(request_xml, encoding="utf-8")
Param(request, D_TYPES.Xml, "request").add_to_param_list(sql_params)
response = await self.call_hq(
"sdk_GetNotifications",
ExecuteStoredProcedure.SqlParameters(sql_params),
use_json=False,
)
if serialize_for_cli:
response = self.serialize_for_cli(response)
return response
[docs]
async def add(self, params: CreateNotificationParams) -> NotificationAddResponse:
"""
Creates a notification row with the provided parameters.
:param params: Instance of CreateNotificationParams dataclass containing the
values of the columns to populate in the Q2_Notifications table.
"""
result = NotificationAddResponse()
sql_params = []
provided_fields = {
key: value for key, value in params.__dict__.items() if value is not None
}
request_xml = tostring(E.Request(E.row(**provided_fields)))
Param(request_xml, D_TYPES.Xml, "requestData").add_to_param_list(sql_params)
response = await self.call_hq(
"Q2_NotificationAddReturnData",
ExecuteStoredProcedure.SqlParameters(sql_params),
use_json=False,
)
if not response:
msg = "Failed to generate notification row"
self.logger.error(msg)
result.error_message = msg
return result
self.logger.info("Successfully generated notification row")
result.success = True
result.id = int(response[0].NotificationID.text)
return result