types: Add SatXMPPClient to core types, add roster and improve type hints.

This commit is contained in:
Goffi 2025-04-11 18:19:28 +02:00
commit 0d40feedc5
2 changed files with 15 additions and 8 deletions

View file

@ -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: ...

View file

@ -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(