import asyncio
from argparse import _SubParsersAction
from datetime import datetime, timezone
from functools import partial
from typing import List, Optional
from lxml.etree import tostring
from lxml.objectify import BoolElement, E, IntElement, StringElement
from q2_sdk.core.dynamic_imports import (
api_ExecuteStoredProcedure as ExecuteStoredProcedure,
)
from q2_sdk.hq.db.admin_logons import AdminLogons
from q2_sdk.hq.db.customer import Customer
from q2_sdk.hq.db.user import User
from q2_sdk.hq.models.hq_params.stored_procedure import Param
from .db_object import DbObject
from .representation_row_base import RepresentationRowBase
D_TYPES = ExecuteStoredProcedure.DataType
[docs]
class CsrNoteRow(RepresentationRowBase):
CsrNoteID: IntElement = "CsrNoteID"
CreatedAt: StringElement = "CreatedAt"
CreatedByAdminUserLogonID: IntElement = "CreatedByAdminUserLogonID"
CustomerID: IntElement = "CustomerID"
UserID: IntElement = "UserID"
Note: StringElement = "Note"
FraudIndicator: BoolElement = "FraudIndicator"
DeletedAt: StringElement = "DeletedAt"
DeletedByAdminUserLoginID: IntElement = "DeletedByAdminUserLoginID"
[docs]
class CsrNote(DbObject):
NAME = "CsrNote"
REPRESENTATION_ROW_CLASS = CsrNoteRow
[docs]
def add_arguments(self, parser: _SubParsersAction):
subparser = parser.add_parser("get_csr_note")
subparser.set_defaults(parser="get")
subparser.set_defaults(func=partial(self.get, serialize_for_cli=True))
subparser.add_argument("-n", "--csr-note-id", help="Q2_CsrNote.CsrNoteID")
subparser.add_argument("-c", "--customer-id", help="Q2_Customer.CustomerID")
subparser.add_argument("-u", "--user-id", help="Q2_User.UserID")
subparser.add_argument(
"-p", "--page-number", help="The page number for the return"
)
subparser.add_argument("-s", "--page-size", help="The page size of the return")
subparser.add_argument(
"-a",
"--created-by-admin-user-logon",
help="Q2_AdminLogons.LoginName. Login name of the admin that is creating the note",
)
subparser.add_argument(
"-d",
"--include-deleted",
action="store_true",
help="include deleted notes",
)
subparser = parser.add_parser("create_csr_note")
subparser.set_defaults(parser="create")
subparser.set_defaults(func=partial(self.create))
subparser.add_argument("customer_id", help="Q2_Customer.CustomerID")
subparser.add_argument("note", help="Q2_CsrNote.Note")
subparser.add_argument("-u", "--user-id", help="Q2_User.UserID")
subparser.add_argument(
"-a",
"--created-by-admin-user-logon",
help="Q2_AdminLogons.LoginName. Login name of the admin that is creating the note",
)
subparser.add_argument(
"-f",
"--fraud_indicator",
action="store_true",
help="Boolean assessor of whether potential fraudulent activity has occurred",
)
subparser = parser.add_parser("delete_csr_note")
subparser.set_defaults(parser="delete")
subparser.set_defaults(func=partial(self.delete))
subparser.add_argument(
"-n", "--csr-note-ids", nargs="+", help="Q2_CsrNote.CsrNoteID"
)
subparser.add_argument(
"-d",
"--deleted-by-admin-user-logon",
help="Q2_AdminLogons.LoginName. Login name of the admin that is deleting the note",
)
[docs]
async def get(
self,
page_number: int = 1,
page_size: int = 100,
csr_note_id: Optional[int] = None,
customer_id: Optional[int] = None,
user_id: Optional[int] = None,
created_by_admin_user_logon: str = "",
include_deleted: bool = False,
serialize_for_cli=False,
) -> List[CsrNoteRow]:
"""
Retrieves CSR Note information by CSR note id, customer id, and/or user id paginated.
:param page_number: the starting point for pagination. Defaults to 1
:param page_size: The number of notes to get per page. Defaults to 100
:param csr note id: The csr note id tied to the note
:param customer_id: The customer id tied to the note
:param user_id: The user id tied to the note
:param created_by_admin_user_logon: The login name of the admin that created the note
:param include_deleted: Boolean indicator of whether deleted entries should be returned. Defaults to False
"""
sql_params = []
assert int(page_number), "page_number must be an integer"
assert int(page_size), "page_size must be an integer"
offset = (int(page_number) - 1) * int(page_size)
assert offset >= 0, "page_number mush be 1 or greater"
Param(offset, D_TYPES.Int, "offset").add_to_param_list(sql_params)
Param(page_size, D_TYPES.Int, "return_count").add_to_param_list(sql_params)
if csr_note_id:
Param(int(csr_note_id), D_TYPES.Int, "csr_note_id").add_to_param_list(
sql_params
)
if customer_id:
Param(int(customer_id), D_TYPES.Int, "customer_id").add_to_param_list(
sql_params
)
if user_id:
Param(int(user_id), D_TYPES.Int, "user_id").add_to_param_list(sql_params)
if include_deleted:
Param(include_deleted, D_TYPES.Bit, "include_deleted").add_to_param_list(
sql_params
)
if created_by_admin_user_logon:
admin_logon_obj = AdminLogons(self.logger, self.hq_credentials)
admin_info = await admin_logon_obj.get(created_by_admin_user_logon)
if not admin_info:
raise ValueError("Invalid Admin User Logon Provided")
admin_logon_id = admin_info[0].AdminUserLogonID.text
Param(
admin_logon_id, D_TYPES.Int, "created_by_admin_user_logon_id"
).add_to_param_list(sql_params)
response = await self.call_hq(
"sdk_GetCsrNote", ExecuteStoredProcedure.SqlParameters(sql_params)
)
if serialize_for_cli:
columns = [
"CsrNoteID",
"UserID",
"CustomerID",
"CreatedByAdminUserLogonID",
"Note",
"DeletedAt",
]
response = self.serialize_for_cli(response, columns)
return response
[docs]
async def create(
self,
customer_id: int,
note: str,
fraud_indicator: bool = False,
user_id: Optional[int] = None,
created_by_admin_user_logon: str = "",
) -> List[CsrNoteRow]:
"""
Creates a CSR Note.
:param customer_id: The customer id tied to the note
:param note: The message to add as a note
:param fraud_indicator: Boolean assessor of whether potential fraudulent activity has occurred
:param user_id: The user id tied to the note. If provided, the note will be created and appear at the user and customer level in Q2 Console
:param created_by_admin_user_logon: The login name of the admin that is creating the note. If not provided, this information
will be retrieved from the HQ Credentials CSR User
"""
asyncio_tasks = []
if user_id:
user_obj = User(self.logger, self.hq_credentials)
asyncio_tasks.append(
asyncio.create_task(user_obj.get(user_id), name="user_info")
)
else:
customer_obj = Customer(self.logger, self.hq_credentials)
asyncio_tasks.append(
asyncio.create_task(customer_obj.get(customer_id), name="customer_info")
)
login_name = (
created_by_admin_user_logon
if created_by_admin_user_logon
else self.hq_credentials.csr_user
)
admin_logon_obj = AdminLogons(self.logger, self.hq_credentials)
asyncio_tasks.append(
asyncio.create_task(admin_logon_obj.get(login_name), name="admin_info")
)
await asyncio.gather(*asyncio_tasks)
for task in asyncio_tasks:
task_name = task.get_name()
result = task.result()
match task_name:
case "user_info":
if not result:
raise ValueError(f"Invalid User ID Provided: {user_id}")
users_customer_id = int(result[0].CustomerID.text)
if customer_id != users_customer_id:
raise ValueError(
f"Provided Customer ID {customer_id} does not match specified User ID {user_id}"
)
case "customer_info":
if not result:
raise ValueError("Invalid Customer ID Provided")
case "admin_info":
if not result:
raise ValueError("Invalid Admin User Logon Provided")
admin_logon_id = result[0].AdminUserLogonID.text
now = datetime.now(timezone.utc)
datetime_format = "%Y-%m-%d %H:%M:%S.%f"
created_at = now.strftime(datetime_format)[:-3]
sql_params = []
Param(created_at, D_TYPES.DateTime, "created_at").add_to_param_list(sql_params)
Param(
admin_logon_id, D_TYPES.Int, "created_by_admin_user_logon_id"
).add_to_param_list(sql_params)
Param(customer_id, D_TYPES.Int, "customer_id").add_to_param_list(sql_params)
Param(note, D_TYPES.VarChar, "note").add_to_param_list(sql_params)
Param(fraud_indicator, D_TYPES.Bit, "fraud_indicator").add_to_param_list(
sql_params
)
if user_id:
Param(user_id, D_TYPES.Int, "user_id").add_to_param_list(sql_params)
response = await self.call_hq(
"sdk_CreateCsrNote", ExecuteStoredProcedure.SqlParameters(sql_params)
)
return response
[docs]
async def delete(
self, csr_note_ids: list[int], deleted_by_admin_user_logon: Optional[str] = None
) -> List[CsrNoteRow]:
"""
Marks a CSR Note as deleted and registers the admin who deleted the note.
:param csr note id: A list of csr note ids to delete
:param deleted_by_admin_user_logon: The login name of the admin that is deleting the note. If not provided, this information
will be retrieved from the HQ Credentials CSR User
"""
assert isinstance(csr_note_ids, list)
# grab admin_logon_id using provided deleted_by_admin_user_logon or CSR user from hq credentials
login_name = (
deleted_by_admin_user_logon
if deleted_by_admin_user_logon
else self.hq_credentials.csr_user
)
admin_logon_obj = AdminLogons(self.logger, self.hq_credentials)
admin_info = await admin_logon_obj.get(login_name)
if not admin_info:
raise ValueError("Invalid Admin User Logon Provided")
admin_logon_id = admin_info[0].AdminUserLogonID.text
# generate a deleted at value
now = datetime.now(timezone.utc)
datetime_format = "%Y-%m-%d %H:%M:%S.%f"
deleted_at = now.strftime(datetime_format)[:-3]
# convert csr user note ids list to xml
ids_to_delete = [E.delete(n=str(id)) for id in csr_note_ids]
request_xml = E.request(
*ids_to_delete,
)
request = tostring(request_xml, encoding="utf-8")
# build sql parameters and fire off request
sql_params = []
Param(request, D_TYPES.Xml, "request").add_to_param_list(sql_params)
Param(deleted_at, D_TYPES.DateTime, "deleted_at").add_to_param_list(sql_params)
Param(
int(admin_logon_id), D_TYPES.Int, "deleted_by_admin_user_login_id"
).add_to_param_list(sql_params)
response = await self.call_hq(
"sdk_DeleteCsrNote",
ExecuteStoredProcedure.SqlParameters(sql_params),
use_json=False,
)
return response