Skip to content
Snippets Groups Projects
Commit 5c9f034e authored by Daniel Brötzmann's avatar Daniel Brötzmann
Browse files

chore: Message: Add type annotations

parent 1b188e22
No related branches found
No related tags found
No related merge requests found
......@@ -14,14 +14,21 @@
# Message handler
from __future__ import annotations
from typing import Any
from typing import Optional
import time
import nbxmpp
from nbxmpp.namespaces import Namespace
from nbxmpp.structs import MessageProperties
from nbxmpp.structs import StanzaHandler
from nbxmpp.util import generate_id
from gajim.common import app
from gajim.common import types
from gajim.common.events import GcMessageReceived
from gajim.common.events import MessageError
from gajim.common.events import MessageReceived
......@@ -34,10 +41,11 @@
from gajim.common.modules.misc import parse_correction
from gajim.common.modules.misc import parse_oob
from gajim.common.modules.misc import parse_xhtml
from gajim.common.structs import OutgoingMessage
class Message(BaseModule):
def __init__(self, con):
def __init__(self, con: types.Client) -> None:
BaseModule.__init__(self, con)
self.handlers = [
......@@ -57,7 +65,11 @@ def __init__(self, con):
self._message_namespaces = set([Namespace.ROSTERX,
Namespace.IBB])
def _check_if_unknown_contact(self, _con, stanza, properties):
def _check_if_unknown_contact(self,
_con: types.xmppClient,
stanza: nbxmpp.Message,
properties: MessageProperties
) -> None:
if (properties.type.is_groupchat or
properties.is_muc_pm or
properties.is_self_message or
......@@ -78,7 +90,11 @@ def _check_if_unknown_contact(self, _con, stanza, properties):
self._log.warning(stanza)
raise nbxmpp.NodeProcessed
def _message_received(self, _con, stanza, properties):
def _message_received(self,
_con: types.xmppClient,
stanza: nbxmpp.Message,
properties: MessageProperties
) -> None:
if (properties.is_mam_message or
properties.is_pubsub or
properties.type.is_error):
......@@ -172,7 +188,7 @@ def _message_received(self, _con, stanza, properties):
if properties.has_security_label:
displaymarking = properties.security_label.displaymarking
event_attr = {
event_attr: dict[str, Any] = {
'conn': self._con,
'stanza': stanza,
'account': self._account,
......@@ -241,7 +257,11 @@ def _message_received(self, _con, stanza, properties):
stanza_id=stanza_id or message_id,
message_id=properties.id)
def _message_error_received(self, _con, _stanza, properties):
def _message_error_received(self,
_con: types.xmppClient,
_stanza: nbxmpp.Message,
properties: MessageProperties
) -> None:
jid = properties.jid
if not properties.is_muc_pm:
jid = jid.new_as_bare()
......@@ -261,7 +281,7 @@ def _message_error_received(self, _con, _stanza, properties):
message_id=properties.id,
error=properties.error))
def _log_muc_message(self, event):
def _log_muc_message(self, event: GcMessageReceived) -> None:
self._check_for_mam_compliance(event.room_jid, event.stanza_id)
if event.msgtxt and event.properties.muc_nickname:
......@@ -276,12 +296,14 @@ def _log_muc_message(self, event):
stanza_id=event.stanza_id,
message_id=event.properties.id)
def _check_for_mam_compliance(self, room_jid, stanza_id):
def _check_for_mam_compliance(self, room_jid: str, stanza_id: str) -> None:
disco_info = app.storage.cache.get_last_disco_info(room_jid)
if stanza_id is None and disco_info.mam_namespace == Namespace.MAM_2:
self._log.warning('%s announces mam:2 without stanza-id', room_jid)
def _get_unique_id(self, properties):
def _get_unique_id(self,
properties: MessageProperties
) -> tuple[Optional[str], Optional[str]]:
if properties.is_self_message:
# Deduplicate self message with message-id
return None, properties.id
......@@ -310,7 +332,7 @@ def _get_unique_id(self, properties):
# stanza-id not added by the archive, ignore it.
return None, None
def build_message_stanza(self, message):
def build_message_stanza(self, message: OutgoingMessage) -> nbxmpp.Message:
own_jid = self._con.get_own_jid()
stanza = nbxmpp.Message(to=message.jid,
......@@ -384,7 +406,7 @@ def build_message_stanza(self, message):
return stanza
def log_message(self, message):
def log_message(self, message: OutgoingMessage) -> None:
if not message.is_loggable:
return
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment