from argparse import _SubParsersAction
from functools import partial
from typing import List
from q2_sdk.core.dynamic_imports import (
api_ExecuteStoredProcedure as ExecuteStoredProcedure,
)
from q2_sdk.hq.table_row import TableRow
from q2_sdk.tools.decorators import dangerous
from .db_object import DbObject
[docs]
class MessageRow(TableRow):
MessageID: int
From: str
To: str
Subject: str
Body: str
ReceivedDate: str
UnreadByCustomer: bool
HasAttachment: bool
ToUser: bool
AllowReplies: bool
[docs]
class MessageThreadRow(TableRow):
Generation: int
MessageID: int
From: str
To: str
Subject: str
Body: str
CreateDate: str
ReceivedDate: str
ExpirationDate: str
ReadDate: str
ParentID: int
ResolvedState: bool
HasAttachment: str
MessageStatus: str
[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