from argparse import _SubParsersAction
from datetime import datetime
from functools import partial
from typing import List
from lxml import etree
from lxml.objectify import IntElement, StringElement
from q2_sdk.core.cli.textui import colored, puts
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.models.demographic import Phone
from .db_object import DbObject
from .representation_row_base import RepresentationRowBase
D_TYPES = ExecuteStoredProcedure.DataType
[docs]
class PhoneNumberRow(RepresentationRowBase):
PhoneID: IntElement = "PhoneID"
CountryID: IntElement = "CountryID"
CityOrAreaCode: StringElement = "CityOrAreaCode"
LocalNumber: StringElement = "LocalNumber"
Extension: StringElement = "Extension"
PhoneType: IntElement = "PhoneType"
UserID: IntElement = "UserID"
CustomerID: IntElement = "CustomerID"
PhoneTypeID: IntElement = "PhoneTypeID"
PhoneTypeName: StringElement = "PhoneTypeName"
IsoCode: IntElement = "IsoCode"
IsoCodeA2: StringElement = "IsoCodeA2"
IsoCodeA3: StringElement = "IsoCodeA3"
CountryName: StringElement = "CountryName"
DialingCode: StringElement = "DialingCode"
SmsDialingCode: StringElement = "SmsDialingCode"
[docs]
class PhoneNumber(DbObject):
NAME = "PhoneNumber"
REPRESENTATION_ROW_CLASS = PhoneNumberRow
[docs]
def add_arguments(self, parser: _SubParsersAction):
subparser = parser.add_parser("get_phone_number")
subparser.set_defaults(parser="get_phone_number")
subparser.set_defaults(func=partial(self.get, serialize_for_cli=True))
subparser.add_argument("user_id", help="Q2_User.UserID")
subparser = parser.add_parser("delete_phone_number")
subparser.set_defaults(parser="delete_phone_number")
subparser.set_defaults(func=partial(self.delete_phone_number_by_id))
subparser.add_argument("user_id", help="Q2_User.UserID")
subparser.add_argument("phone_id", help="Q2_PhoneNumber.PhoneID")
subparser = parser.add_parser("add_phone_number")
subparser.set_defaults(parser="add_phone_number")
subparser.set_defaults(func=partial(self.add_phone_number))
subparser.add_argument("country_id", help="Q2_Country.CountryID")
subparser.add_argument(
"city_or_area_code", help="Q2_PhoneNumber.CityOrAreaCode"
)
subparser.add_argument("local_number", help="Q2_PhoneNumber.LocalNumber")
subparser.add_argument("phone_type", help="Q2_PhoneNumber.PhoneType")
subparser.add_argument("-u", "--user-id", help="Q2_PhoneNumber.PhoneType")
subparser.add_argument("-c", "--customer-id", help="Q2_PhoneNumber.PhoneType")
[docs]
async def get(self, user_id: int, serialize_for_cli=False) -> List[PhoneNumberRow]:
try:
int(user_id)
except ValueError as err:
error_msg = "Please supply a valid user ID"
puts(colored.red(error_msg))
raise ValueError(error_msg) from err
response = await self.call_hq(
"sdk_GetPhoneNumber",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int, "user_id", user_id
)
]),
)
if serialize_for_cli:
columns = [
"PhoneID",
"CityOrAreaCode",
"LocalNumber",
"PhoneTypeName",
"CountryName",
"Extension",
]
response = self.serialize_for_cli(response, columns)
return response
[docs]
async def get_customer_phone_numbers(
self, customer_id: int
) -> List[PhoneNumberRow]:
try:
int(customer_id)
except ValueError as err:
error_msg = "Please supply a valid customer ID"
raise ValueError(error_msg) from err
response = await self.call_hq(
"sdk_GetPhoneNumber",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int, "customer_id", customer_id
)
]),
)
return response
[docs]
async def delete_phone_number_by_id(
self, user_id: int, phone_id: int, date_time: datetime = None
):
# note: datetime variable is unused by storedproc
if not date_time:
date_time = datetime.now().isoformat()
try:
phone_id = int(phone_id)
user_id = int(user_id)
except ValueError as err:
error_msg = "phone_id and user_id must be integers"
raise ValueError(error_msg) from err
params = []
Param(phone_id, D_TYPES.Int, "Id").add_to_param_list(params)
Param(user_id, D_TYPES.Int, "ParentId").add_to_param_list(params)
Param(date_time, D_TYPES.DateTime, "DateTime").add_to_param_list(params)
await self.call_hq(
stored_proc_short_name="Q2_PhoneNumber_Delete_EndUser",
sql_parameters=ExecuteStoredProcedure.SqlParameters(params),
)
[docs]
async def add_phone_number(
self,
country_id,
city_or_area_code,
local_number,
phone_type,
user_id=None,
customer_id=None,
extension=None,
):
try:
int(local_number)
except ValueError as e:
raise ValueError("Please provide a valid local number") from e
param_list = []
Param(country_id, D_TYPES.Int, "country_id").add_to_param_list(param_list)
Param(
city_or_area_code, D_TYPES.VarChar, "city_or_area_code"
).add_to_param_list(param_list)
Param(local_number, D_TYPES.VarChar, "local_number").add_to_param_list(
param_list
)
Param(phone_type, D_TYPES.Int, "phone_type").add_to_param_list(param_list)
Param(user_id, D_TYPES.Int, "user_id").add_to_param_list(param_list)
Param(customer_id, D_TYPES.Int, "customer_id").add_to_param_list(param_list)
Param(extension, D_TYPES.VarChar, "extension").add_to_param_list(param_list)
await self.call_hq(
stored_proc_short_name="sdk_AddPhoneNumber",
sql_parameters=ExecuteStoredProcedure.SqlParameters(param_list),
)
[docs]
async def update_customer_phones(
self, customer_id: int, phones: list[Phone]
) -> list[str]:
"""
:param customer_id: the CustomerID of the customer you would like to update
:param phones: a list of phone objects containing the customer's new phone information
:return list containing success or failure message string
"""
assert isinstance(customer_id, int), "Please provide a valid customer_id"
assert all([isinstance(phone, Phone) for phone in phones]), (
"Please provide a valid list of phones"
)
phone_node = etree.Element("Customer", customer_id=str(customer_id))
for phone in phones:
etree.SubElement(
phone_node,
"Phone",
country=phone.country,
area=phone.area_code,
number=phone.phone_number,
phone_type=phone.type,
extension=phone.extension,
)
xml_request = etree.tostring(phone_node).decode()
params = []
Param(xml_request, D_TYPES.Xml, "custPhoneXML").add_to_param_list(params)
response = await self.call_hq(
"sdk_UpdateCustomerPhones", ExecuteStoredProcedure.SqlParameters(params)
)
return response