Source code for q2_sdk.hq.db.message

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