from argparse import _SubParsersAction
from datetime import datetime
from functools import partial
from typing import List, Optional
from q2_sdk.core.dynamic_imports import (
api_ExecuteStoredProcedure as ExecuteStoredProcedure,
)
from q2_sdk.hq.models.alerts import DateAlertDefinition
from q2_sdk.hq.models.hq_params.stored_procedure import ParamsBuilder
from q2_sdk.hq.table_row import TableRow
from .db_object import DbObject
D_TYPES = ExecuteStoredProcedure.DataType
[docs]
class DateAlertRow(TableRow):
AlertDefinitionID: int
AlertTypeID: int
UserID: int
Enabled: bool
SendSecureMessage: bool
OneTimeAlert: bool
NotificationTypeID: int
NotificationTime: str
Operand: str
ComparisonValue: str
LastAlerted: str
CountryID: int
CityOrAreaCode: str
LocalNumber: str
EmailAddress: str
CreateDate: str
AlertDateTypeID: str
UserText: str
[docs]
class DateAlert(DbObject):
# GET_BY_NAME_KEY = "column in the db response"
NAME = "DateAlert"
REPRESENTATION_ROW_CLASS = DateAlertRow
[docs]
def add_arguments(self, parser: _SubParsersAction):
subparser = parser.add_parser("get_date_alert")
subparser.set_defaults(parser="get")
subparser.set_defaults(func=partial(self.get, serialize_for_cli=True))
subparser.add_argument("user_id", help="Q2_User.UserID", type=int)
[docs]
async def get(self, user_id: int, serialize_for_cli=False) -> List[DateAlertRow]:
assert isinstance(user_id, int), "Please supply a valid user id"
response = await self.call_hq(
"sdk_GetDateAlert",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int, "user_id", user_id
)
]),
)
if serialize_for_cli:
columns = [
"AlertDefinitionID",
"UserID",
"Enabled",
"OneTimeAlert",
"NotificationTypeID",
"Operand",
"ComparisonValue",
"LastAlerted",
]
response = self.serialize_for_cli(response, columns)
return response
[docs]
async def create(
self, alert_info: DateAlertDefinition, create_date: Optional[str] = None
):
assert isinstance(alert_info, DateAlertDefinition), (
"Please supply a valid DateAlert object"
)
time_stamp = (
create_date
if create_date
else datetime.now().strftime("%Y-%m-%dT%H:%M:%S.%f")
)
params_builder = self._build_parameters(alert_info)
params_builder.add_param(D_TYPES.DateTime, "CreateDate", time_stamp)
response = await self.call_hq(
"Q2_AlertAddDate",
ExecuteStoredProcedure.SqlParameters(params_builder.build()),
)
return response
[docs]
async def update(self, alert_info: DateAlertDefinition, alert_definition_id: int):
assert isinstance(alert_info, DateAlertDefinition), (
"Please supply a valid DateAlert object"
)
params_builder = self._build_parameters(alert_info)
params_builder.add_param(D_TYPES.Bit, "Enabled", alert_info.enabled)
params_builder.add_param(D_TYPES.Int, "AlertDefinitionID", alert_definition_id)
response = await self.call_hq(
"Q2_AlertUpdateDate",
ExecuteStoredProcedure.SqlParameters(params_builder.build()),
)
return response
@staticmethod
def _build_parameters(alert_info: DateAlertDefinition):
params_builder = (
ParamsBuilder()
.add_param(D_TYPES.Int, "UserID", alert_info.user_id)
.add_param(D_TYPES.Bit, "SendSecureMessage", alert_info.send_secure_message)
.add_param(D_TYPES.Bit, "OneTimeAlert", alert_info.one_time_alert)
.add_param(D_TYPES.VarChar, "Operand", alert_info.operand.value)
.add_param(D_TYPES.VarChar, "ComparisonValue", alert_info.comparison_value)
.add_param(D_TYPES.Int, "AlertDateTypeID", alert_info.alert_date_type.value)
.add_param(
D_TYPES.Int,
"NotificationTypeID",
alert_info.notification_type.value,
optional=True,
)
.add_param(
D_TYPES.DateTime,
"NotificationTime",
alert_info.notification_time,
optional=True,
)
.add_param(
D_TYPES.VarChar, "EmailAddress", alert_info.email_address, optional=True
)
.add_param(
D_TYPES.VarChar,
"LocalNumber",
alert_info.local_phone_number,
optional=True,
)
.add_param(
D_TYPES.VarChar,
"CityOrAreaCode",
alert_info.city_or_area_code,
optional=True,
)
.add_param(D_TYPES.Int, "CountryID", alert_info.country_id, optional=True)
.add_param(D_TYPES.VarChar, "UserText", alert_info.user_text, optional=True)
)
return params_builder