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 Param
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")
)
parameters = self._build_parameters(alert_info)
Param(time_stamp, D_TYPES.DateTime, "CreateDate").add_to_param_list(parameters)
response = await self.call_hq(
"Q2_AlertAddDate", ExecuteStoredProcedure.SqlParameters(parameters)
)
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"
)
parameters = self._build_parameters(alert_info)
Param(alert_info.enabled, D_TYPES.Bit, "Enabled").add_to_param_list(parameters)
Param(alert_definition_id, D_TYPES.Int, "AlertDefinitionID").add_to_param_list(
parameters
)
response = await self.call_hq(
"Q2_AlertUpdateDate", ExecuteStoredProcedure.SqlParameters(parameters)
)
return response
@staticmethod
def _build_parameters(alert_info: DateAlertDefinition):
parameters = []
possible_params = [
Param(alert_info.user_id, D_TYPES.Int, "UserID"),
Param(alert_info.send_secure_message, D_TYPES.Bit, "SendSecureMessage"),
Param(alert_info.one_time_alert, D_TYPES.Bit, "OneTimeAlert"),
Param(alert_info.operand.value, D_TYPES.VarChar, "Operand"),
Param(alert_info.comparison_value, D_TYPES.VarChar, "ComparisonValue"),
Param(alert_info.alert_date_type.value, D_TYPES.Int, "AlertDateTypeID"),
Param(
alert_info.notification_type.value,
D_TYPES.Int,
"NotificationTypeID",
True,
),
Param(
alert_info.notification_time, D_TYPES.DateTime, "NotificationTime", True
),
Param(alert_info.email_address, D_TYPES.VarChar, "EmailAddress", True),
Param(alert_info.local_phone_number, D_TYPES.VarChar, "LocalNumber", True),
Param(
alert_info.city_or_area_code, D_TYPES.VarChar, "CityOrAreaCode", True
),
Param(alert_info.country_id, D_TYPES.Int, "CountryID", True),
Param(alert_info.user_text, D_TYPES.VarChar, "UserText", True),
]
for item in possible_params:
item.add_to_param_list(parameters)
return parameters