Source code for spade.message

import logging
from typing import Optional, Dict, Type

import aioxmpp
import aioxmpp.forms.xso as forms_xso

SPADE_X_METADATA = "spade:x:metadata"

logger = logging.getLogger("spade.Message")

[docs]class MessageBase(object): """ """ def __init__( self, to: Optional[str] = None, sender: Optional[str] = None, body: Optional[str] = None, thread: Optional[str] = None, metadata: Optional[Dict[str, str]] = None, ): self.sent = False self._to, self._sender, self._body, self._thread = None, None, None, None = to self.sender = sender self.body = body self.thread = thread if metadata is None: self.metadata = {} else: for key, value in metadata.items(): if not isinstance(key, str) or not isinstance(value, str): raise TypeError("Key and Value of metadata MUST be strings") self.metadata = metadata
[docs] @classmethod def from_node(cls, node: aioxmpp.Message) -> Type['MessageBase']: """ Creates a new spade.message.Message from an aixoxmpp.stanza.Message Args: node (aioxmpp.stanza.Message): an aioxmpp Message Returns: spade.message.Message: a new spade Message """ if not isinstance(node, aioxmpp.stanza.Message): raise AttributeError("node must be a aioxmpp.stanza.Message instance") msg = cls() msg._to = msg._sender = node.from_ if None in node.body: msg.body = node.body[None] else: for key in node.body.keys(): msg.body = node.body[key] break for data in node.xep0004_data: if data.title == SPADE_X_METADATA: for field in data.fields: if field.var != "_thread_node": msg.set_metadata(field.var, field.values[0]) else: msg.thread = field.values[0] return msg
@property def to(self) -> aioxmpp.JID: """ Gets the jid of the receiver. Returns: aioxmpp.JID: jid of the receiver """ return self._to @to.setter def to(self, jid: str) -> None: """ Set jid of the receiver. Args: jid (str): the jid of the receiver. """ if jid is not None and not isinstance(jid, str): raise TypeError("'to' MUST be a string") self._to = aioxmpp.JID.fromstr(jid) if jid is not None else None @property def sender(self) -> aioxmpp.JID: """ Get jid of the sender Returns: aioxmpp.JID: jid of the sender """ return self._sender @sender.setter def sender(self, jid: str) -> None: """ Set jid of the sender Args: jid (str): jid of the sender """ if jid is not None and not isinstance(jid, str): raise TypeError("'sender' MUST be a string") self._sender = aioxmpp.JID.fromstr(jid) if jid is not None else None @property def body(self) -> str: """ Get body of the message Returns: str: the body of the message """ return self._body @body.setter def body(self, body: str) -> None: """ Set body of the message Args: body (str): The body of the message """ if body is not None and not isinstance(body, str): raise TypeError("'body' MUST be a string") self._body = body @property def thread(self) -> str: """ Get Thread of the message Returns: str: thread id """ return self._thread @thread.setter def thread(self, value: str) -> None: """ Set thread id of the message Args: value (str): the thread id """ if value is not None and not isinstance(value, str): raise TypeError("'thread' MUST be a string") self._thread = value
[docs] def set_metadata(self, key: str, value: str) -> None: """ Add a new metadata to the message Args: key (str): name of the metadata value (str): value of the metadata """ if not isinstance(key, str) or not isinstance(value, str): raise TypeError("'key' and 'value' of metadata MUST be strings") self.metadata[key] = value
[docs] def get_metadata(self, key: str) -> str: """ Get the value of a metadata. Returns None if metadata does not exist. Args: key (str): name of the metadata Returns: str: the value of the metadata (or None) """ return self.metadata[key] if key in self.metadata else None
[docs] def match(self, message: Type['MessageBase']) -> bool: """ Returns wether a message matches with this message or not. The message can be a Message object or a Template object. Args: message (spade.message.Message): the message to match to Returns: bool: wether the message matches or not """ if and != return False if self.sender and message.sender != self.sender: return False if self.body and message.body != self.body: return False if self.thread and message.thread != self.thread: return False for key, value in self.metadata.items(): if message.get_metadata(key) != value: return False logger.debug(f"message matched {self} == {message}") return True
@property def id(self) -> int: """ """ return id(self) def __eq__(self, other: Type['MessageBase']): return self.match(other)
[docs]class Message(MessageBase): """ """
[docs] def make_reply(self) -> 'Message': """ Creates a copy of the message, exchanging sender and receiver Returns: spade.message.Message: a new message with exchanged sender and receiver """ return Message( to=str(self.sender), sender=str(, body=self.body, thread=self.thread, metadata=self.metadata, )
[docs] def prepare(self) -> aioxmpp.Message: """ Returns an aioxmpp.stanza.Message built from the Message and prepared to be sent. Returns: aioxmpp.stanza.Message: the message prepared to be sent """ msg = aioxmpp.stanza.Message(, from_=self.sender, type_=aioxmpp.MessageType.CHAT, ) msg.body[None] = self.body # Send metadata using xep-0004: Data Forms ( if len(self.metadata): data = forms_xso.Data(type_=forms_xso.DataType.FORM) for name, value in self.metadata.items(): data.fields.append( forms_xso.Field( var=name, type_=forms_xso.FieldType.TEXT_SINGLE, values=[value], ) ) if self.thread: data.fields.append( forms_xso.Field( var="_thread_node", type_=forms_xso.FieldType.TEXT_SINGLE, values=[self.thread], ) ) data.title = SPADE_X_METADATA msg.xep0004_data = [data] return msg
def __str__(self) -> str: s = f'<message to="{}" from="{self.sender}" thread="{self.thread}" metadata={self.metadata}>' if self.body: s += "\n" + self.body + "\n" s += "</message>" return s