diff --git a/libervia/backend/core/core_types.py b/libervia/backend/core/core_types.py index a59c3150..4e07ec05 100644 --- a/libervia/backend/core/core_types.py +++ b/libervia/backend/core/core_types.py @@ -18,7 +18,7 @@ from abc import ABC, abstractmethod from collections import namedtuple -from typing import Any +from typing import TYPE_CHECKING, Any from twisted.internet import defer from twisted.python import failure from typing_extensions import TypedDict @@ -30,6 +30,9 @@ from wokkel import disco from libervia.backend.core import exceptions +if TYPE_CHECKING: + from libervia.backend.core.xmpp import LiberviaRosterProtocol + EncryptionPlugin = namedtuple( "EncryptionPlugin", ("instance", "name", "namespace", "priority", "directed") @@ -261,6 +264,10 @@ class SatXMPPEntity(ABC): raise NotImplementedError +class SatXMPPClient(SatXMPPEntity): + roster: "LiberviaRosterProtocol" + + class SatXMPPComponent(SatXMPPEntity): def is_local(self, jid_: t_jid.JID) -> bool: ... diff --git a/libervia/backend/core/xmpp.py b/libervia/backend/core/xmpp.py index 95982a07..bfeaf275 100644 --- a/libervia/backend/core/xmpp.py +++ b/libervia/backend/core/xmpp.py @@ -941,7 +941,7 @@ ExtraDict = dict # TODO @implementer(iwokkel.IDisco) -class SatXMPPClient(SatXMPPEntity, wokkel_client.XMPPClient): +class SatXMPPClient(SatXMPPEntity, core_types.SatXMPPClient, wokkel_client.XMPPClient): trigger_suffix = "" is_component = False @@ -1553,8 +1553,8 @@ class LiberviaRosterProtocol(xmppim.RosterClientProtocol): self.host = host self.got_roster = defer.Deferred() # called when roster is received and ready # XXX: the two following dicts keep a local copy of the roster - self._jids = {} # map from jids to RosterItem: key=jid value=RosterItem - self._groups = {} # map from groups to jids: key=group value=set of jids + self._jids: dict[jid.JID, xmppim.RosterItem] = {} + self._groups: dict[str, set[jid.JID]] = {} # map from groups to jids def __contains__(self, entity_jid): return self.is_jid_in_roster(entity_jid) @@ -1773,11 +1773,11 @@ class LiberviaRosterProtocol(xmppim.RosterClientProtocol): # then we send the bridge signal self.host.bridge.contact_deleted(entity.full(), self.parent.profile) - def get_groups(self): + def get_groups(self) -> list[str]: """Return a list of groups""" return list(self._groups.keys()) - def get_item(self, entity_jid): + def get_item(self, entity_jid) -> xmppim.RosterItem | None: """Return RosterItem for a given jid @param entity_jid(jid.JID): jid of the contact @@ -1786,11 +1786,11 @@ class LiberviaRosterProtocol(xmppim.RosterClientProtocol): """ return self._jids.get(entity_jid, None) - def get_jids(self): + def get_jids(self) -> list[jid.JID]: """Return all jids of the roster""" return list(self._jids.keys()) - def is_jid_in_roster(self, entity_jid): + def is_jid_in_roster(self, entity_jid) -> bool: """Return True if jid is in roster""" if not isinstance(entity_jid, jid.JID): raise exceptions.InternalError(