from argparse import _SubParsersAction
from functools import partial
from typing import List
from lxml.objectify import IntElement, StringElement, BoolElement
from q2_sdk.core.dynamic_imports import (
api_ExecuteStoredProcedure as ExecuteStoredProcedure,
)
from q2_sdk.tools.decorators import dangerous
from .db_object import DbObject
from .representation_row_base import RepresentationRowBase
[docs]
class MessageRow(RepresentationRowBase):
MessageID: IntElement = "MessageID"
From: StringElement = "From"
To: StringElement = "To"
Subject: StringElement = "Subject"
Body: StringElement = "Body"
ReceivedDate: StringElement = "ReceivedDate"
UnreadByCustomer: BoolElement = "UnreadByCustomer"
HasAttachment: BoolElement = "HasAttachment"
ToUser: BoolElement = "ToUser"
AllowReplies: BoolElement = "AllowReplies"
[docs]
class MessageThreadRow(RepresentationRowBase):
Generation: IntElement = "Generation"
MessageID: IntElement = "MessageID"
From: StringElement = "From"
To: StringElement = "To"
Subject: StringElement = "Subject"
Body: StringElement = "Body"
CreateDate: StringElement = "CreateDate"
ReceivedDate: StringElement = "ReceivedDate"
ExpirationDate: StringElement = "ExpirationDate"
ReadDate: StringElement = "ReadDate"
ParentID: IntElement = "ParentID"
ResolvedState: BoolElement = "ResolvedState"
HasAttachment: StringElement = "HasAttachment"
MessageStatus: StringElement = "MessageStatus"
[docs]
class Message(DbObject):
NAME = "Message"
REPRESENTATION_ROW_CLASS = MessageRow
[docs]
def add_arguments(self, parser: _SubParsersAction):
subparser = parser.add_parser("get_message")
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")
subparser = parser.add_parser("get_message_thread")
subparser.set_defaults(parser="get_message_thread")
subparser.set_defaults(
func=partial(self.get_message_thread, serialize_for_cli=True)
)
subparser.add_argument("message_id", help="Q2_Message.MessageID")
[docs]
async def get(self, user_id: int, serialize_for_cli=False) -> List[MessageRow]:
try:
int(user_id)
except ValueError as err:
raise AssertionError("Not a valid user id") from err
response = await self.call_hq(
"Q2_MessageConversationList",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int, "UserID", user_id
)
]),
)
if serialize_for_cli:
columns = ["MessageID", "From", "To", "Subject"]
response = self.serialize_for_cli(response, columns)
return response
@dangerous(
"This has the potential to negatively impact the database. It is not advised to use this in loops"
)
async def get_message_thread(
self,
message_id: int,
relation_code: int = None,
include_forward_messages=False,
serialize_for_cli=False,
) -> List[MessageThreadRow]:
"""
Returns a message thread for a given message id
relation_code values
x01 = direct line ancestors
x02 = direct line and sibling ancestors
x04 = direct line children
x08 = direct line and sibling children
valid combinations 0,1,3,4,5,7,15
"""
try:
int(message_id)
except ValueError as err:
raise AssertionError("Not a valid message id") from err
message_relation_code = 1
if relation_code:
message_relation_code = relation_code
response = await self.call_hq(
"Q2_spGetMessageTree",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int, "messageID", message_id
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int,
"relation",
message_relation_code,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Bit,
"includeForwardedMessages",
include_forward_messages,
),
]),
representation_class_override=MessageThreadRow,
)
if serialize_for_cli:
columns = ["MessageID", "From", "To", "Subject", "ParentID"]
response = self.serialize_for_cli(response, columns)
return response