import asyncio
from argparse import _SubParsersAction
from functools import partial
from typing import List
from lxml.objectify import IntElement, StringElement
from q2_sdk.core.dynamic_imports import (
api_ExecuteStoredProcedure as ExecuteStoredProcedure,
)
from q2_sdk.core.exceptions import DatabaseDataError
from q2_sdk.tools.decorators import dev_only
from .db_object import DbObject
from .email import Email
from .phone_number import PhoneNumber
from .push_notification_targets import PushNotificationTargets
from .representation_row_base import RepresentationRowBase
from .user import User
[docs]
class SecAlertUserPrefRow(RepresentationRowBase):
SecAlertUserPrefID: IntElement = "SecAlertUserPrefID"
UserID: IntElement = "UserID"
EmailID: IntElement = "EmailID"
SmsID: IntElement = "SmsID"
PhoneID: IntElement = "PhoneID"
PushTargetID: IntElement = "PushTargetID"
FirstName: StringElement = "FirstName"
LastName: StringElement = "LastName"
EmailAddress: StringElement = "EmailAddress"
PhoneNumber: StringElement = "PhoneNumber"
CountryCodePhoneNumber: StringElement = "CountryCodePhoneNumber"
AreaCodePhoneNumber: StringElement = "AreaCodePhoneNumber"
SmsNumber: StringElement = "SmsNumber"
CountryCodeSmsNumber: StringElement = "CountryCodeSmsNumber"
AreaCodeSmsNumber: StringElement = "AreaCodeSmsNumber"
Nickname: StringElement = "Nickname"
[docs]
class SecAlertUserPref(DbObject):
NAME = "SecAlertUserPref"
REPRESENTATION_ROW_CLASS = SecAlertUserPrefRow
[docs]
def add_arguments(self, parser: _SubParsersAction):
subparser = parser.add_parser("get_sec_alert_user_pref")
subparser.set_defaults(parser="get")
subparser.set_defaults(func=partial(self.get, serialize_for_cli=True))
subparser.add_argument("user_login_name", help="Q2_UserLogon.LoginName")
subparser = parser.add_parser("add_sec_alert_user_pref")
subparser.set_defaults(parser="add")
subparser.set_defaults(func=partial(self.add))
subparser.add_argument("user_id", type=int, help="Q2_User.UserID")
subparser.add_argument("-e", "--email-address", help="Q2_Email.EmailAddress")
subparser.add_argument("-p", "--phone-id", help="Q2_SecAlertUserPref.PhoneID")
subparser.add_argument("-s", "--sms-id", help="Q2_SecAlertUserPref.SmsID")
subparser.add_argument(
"-pt", "--push-target-id", help="Q2_SecAlertUserPref.PushTargetID"
)
subparser = parser.add_parser("update_sec_alert_user_pref")
subparser.set_defaults(parser="update")
subparser.set_defaults(func=partial(self.update))
subparser.add_argument("-u", "--user-id", help="Q2_SecAlertUserPref.UserID")
subparser.add_argument("-e", "--email-id", help="Q2_SecAlertUserPref.EmailID")
subparser.add_argument(
"-ea", "--email-address", help="Q2_SecAlertUserPref.EmailAddress"
)
subparser.add_argument("-p", "--phone-id", help="Q2_SecAlertUserPref.PhoneID")
subparser.add_argument(
"-pc", "--phone-country-id", help="Q2_PhoneNumber.CountryID"
)
subparser.add_argument(
"-pa", "--phone-city-or-area-code", help="Q2_PhoneNumber.CityOrAreaCode"
)
subparser.add_argument(
"-pl", "--phone-local-number", help="Q2_PhoneNumber.LocalNumber"
)
subparser.add_argument(
"-pe", "--phone-extension", help="Q2_PhoneNumber.Extension"
)
subparser.add_argument("-s", "--sms-id", help="Q2_PhoneNumber.PhoneID")
subparser.add_argument(
"-sc", "--sms-country-id", help="Q2_PhoneNumber.CountryID"
)
subparser.add_argument(
"-sa", "--sms-city-or-area-code", help="Q2_PhoneNumber.CityOrAreaCode"
)
subparser.add_argument(
"-sl", "--sms-local-number", help="Q2_PhoneNumber.LocalNumber"
)
subparser.add_argument(
"-se", "--sms-extension", help="Q2_PhoneNumber.Extension"
)
subparser.add_argument(
"-pt", "--push-target-id", help="Q2_SecAlertUserPref.PushTargetID"
)
subparser = parser.add_parser("remove_sec_alert_user_pref")
subparser.set_defaults(parser="remove")
subparser.set_defaults(func=partial(self.remove))
subparser.add_argument(
"sec_alert_user_pref_id", help="Q2_SecAlertUserPref.SecAlertUserPrefID"
)
[docs]
async def get(
self, user_login_name: str = None, serialize_for_cli=False
) -> List[SecAlertUserPrefRow]:
response = await self.call_hq(
"sdk_GetSecAlertUserPref",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"UserLogonName",
user_login_name,
)
]),
)
if serialize_for_cli:
columns = [
"SecAlertUserPrefID",
"FirstName",
"LastName",
"EmailAddress",
"PhoneNumber",
"CountryCodePhoneNumber",
"AreaCodePhoneNumber",
"SmsNumber",
"CountryCodeSmsNumber",
"AreaCodeSmsNumber",
"Nickname",
]
response = self.serialize_for_cli(
response, columns, fields_to_truncate=["EmailAddress"]
)
return response
[docs]
async def add(
self,
user_id: int,
email_address: str = None,
phone_id: int = None,
sms_id: int = None,
push_target_id: int = None,
):
user = User(self.logger, hq_credentials=self.hq_credentials, ret_table_obj=True)
user_row = await user.get(user_id)
if not user_row:
raise DatabaseDataError(
f"No User with UserID '{user_id}' is present in the database"
)
email = Email(
self.logger, hq_credentials=self.hq_credentials, ret_table_obj=True
)
phone = PhoneNumber(
self.logger, hq_credentials=self.hq_credentials, ret_table_obj=True
)
push_target = PushNotificationTargets(
self.logger, hq_credentials=self.hq_credentials, ret_table_obj=True
)
email_rows, phone_rows, push_target_rows = await asyncio.gather(
email.get_by_user(user_id), phone.get(user_id), push_target.get(user_id)
)
email_row = [
er
for er in email_rows
if email_address and er.EmailAddress == email_address
]
phone_row = [
phone for phone in phone_rows if phone_id and phone.PhoneID == int(phone_id)
]
sms_row = [sms for sms in phone_rows if sms_id and sms.PhoneID == int(sms_id)]
push_target_row = [
push_target
for push_target in push_target_rows
if push_target_id and push_target.PushTargetID == int(push_target_id)
]
if email_address and not email_row:
raise EmailUnrelatedToUserError(
f"User with UserID '{user_id}' does not have email address '{email_address}'"
)
if phone_id and not phone_row:
raise PhoneUnrelatedToUserError(
f"User with username '{user_id}' does not have phone number with phone id '{phone_id}'"
)
if sms_id and not sms_row:
raise SmsUnrelatedToUserError(
f"User with username '{user_id}' does not have phone number with phone id '{sms_id}'"
)
if push_target_id and not push_target_row:
raise PushTargetUnrelatedToUserError(
f"User with username '{user_id}' does not have Push Notification Target '{push_target_id}'"
)
email_id = email_row[0].EmailID if email_row else None
s_param = ExecuteStoredProcedure.SqlParam
d_type = ExecuteStoredProcedure.DataType
response = await self.call_hq(
"sdk_AddSecAlertUserPref",
ExecuteStoredProcedure.SqlParameters([
s_param(d_type.Int, "UserID", user_id),
s_param(d_type.Int, "EmailID", email_id),
s_param(d_type.Int, "PhoneID", phone_id),
s_param(d_type.Int, "SmsID", sms_id),
s_param(d_type.Int, "PushTargetID", push_target_id),
]),
)
return response
[docs]
async def update(
self,
user_id=None,
email_id=None,
email_address=None,
phone_id=None,
phone_country_id=None,
phone_city_or_area_code=None,
phone_local_number=None,
phone_extension=None,
sms_id=None,
sms_country_id=None,
sms_city_or_area_code=None,
sms_local_number=None,
sms_extension=None,
push_target_id=None,
):
s_param = ExecuteStoredProcedure.SqlParam
d_type = ExecuteStoredProcedure.DataType
params = [
s_param(d_type.Int, "UserID", user_id),
s_param(d_type.Int, "EmailID", email_id),
s_param(d_type.VarChar, "EmailAddress", email_address),
s_param(d_type.Int, "PhoneID", phone_id),
s_param(d_type.Int, "PhoneCountryID", phone_country_id),
s_param(d_type.VarChar, "PhoneCityOrAreaCode", phone_city_or_area_code),
s_param(d_type.VarChar, "PhoneLocalNumber", phone_local_number),
s_param(d_type.VarChar, "PhoneExtension", phone_extension),
s_param(d_type.Int, "SmsID", sms_id),
s_param(d_type.Int, "SmsCountryID", sms_country_id),
s_param(d_type.VarChar, "SmsCityOrAreaCode", sms_city_or_area_code),
s_param(d_type.VarChar, "SmsLocalNumber", sms_local_number),
s_param(d_type.VarChar, "SmsExtension", sms_extension),
s_param(d_type.Int, "PushTargetID", push_target_id),
]
sql_params = []
for item in params:
sql_params.append(item)
response = await self.call_hq(
"Q2_SecAlertUserPrefUpdate",
ExecuteStoredProcedure.SqlParameters(sql_params),
)
return response
[docs]
@dev_only
async def remove(self, sec_alert_user_pref_id):
"""Note: this only works in the dev environment"""
response = await self.call_hq(
"sdk_RemoveSecAlertUserPref",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int,
"SecAlertUserPrefID",
sec_alert_user_pref_id,
)
]),
)
return response