summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authormathieui <mathieui@mathieui.net>2021-03-04 22:14:59 +0100
committermathieui <mathieui@mathieui.net>2021-03-04 22:14:59 +0100
commitc1a598c34be8ca5b85b69ce9a442363686261e35 (patch)
tree837fe67052d81b288cec8922b8e3e130fd07eae1
parent2314ee422ae815faea4b437588241f220ff51d71 (diff)
parent49416dacbcf3c4070c16a29370f76dfef6dfad46 (diff)
downloadslixmpp-c1a598c34be8ca5b85b69ce9a442363686261e35.tar.gz
slixmpp-c1a598c34be8ca5b85b69ce9a442363686261e35.tar.bz2
slixmpp-c1a598c34be8ca5b85b69ce9a442363686261e35.tar.xz
slixmpp-c1a598c34be8ca5b85b69ce9a442363686261e35.zip
Merge branch 'xep-0030-typing-and-co' into 'master'
XEP-0030: add typing and more docs See merge request poezio/slixmpp!143
-rw-r--r--docs/api/plugins/xep_0030.rst17
-rw-r--r--slixmpp/plugins/xep_0030/disco.py234
-rw-r--r--slixmpp/plugins/xep_0030/stanza/info.py74
-rw-r--r--slixmpp/plugins/xep_0030/stanza/items.py74
-rw-r--r--slixmpp/plugins/xep_0030/static.py300
-rw-r--r--slixmpp/types.py10
6 files changed, 424 insertions, 285 deletions
diff --git a/docs/api/plugins/xep_0030.rst b/docs/api/plugins/xep_0030.rst
index 81499dba..188ede45 100644
--- a/docs/api/plugins/xep_0030.rst
+++ b/docs/api/plugins/xep_0030.rst
@@ -9,14 +9,31 @@ XEP-0030: Service Discovery
:exclude-members: session_bind, plugin_init, plugin_end
+.. _api-0030:
+
+Internal API Methods
+--------------------
+
+All ``api`` operations supported by the 0030 plugin are implemented as part of
+the :class:`~.StaticDisco` class which implement an in-memory cache for disco
+info and items.
+
+
+.. automodule:: slixmpp.plugins.xep_0030.static
+ :members:
+ :member-order: bysource
+
+
Stanza elements
---------------
.. automodule:: slixmpp.plugins.xep_0030.stanza.info
:members:
+ :member-order: bysource
:undoc-members:
.. automodule:: slixmpp.plugins.xep_0030.stanza.items
:members:
+ :member-order: bysource
:undoc-members:
diff --git a/slixmpp/plugins/xep_0030/disco.py b/slixmpp/plugins/xep_0030/disco.py
index 6aeadc84..08dedbce 100644
--- a/slixmpp/plugins/xep_0030/disco.py
+++ b/slixmpp/plugins/xep_0030/disco.py
@@ -7,14 +7,20 @@ import asyncio
import logging
from asyncio import Future
-from typing import Optional, Callable
-
-from slixmpp import Iq
-from slixmpp import future_wrapper
+from typing import (
+ Optional,
+ Callable,
+ List,
+ Union,
+)
+
+from slixmpp import JID
+from slixmpp.stanza import Iq
+from slixmpp.types import OptJid
from slixmpp.plugins import BasePlugin
-from slixmpp.xmlstream.handler import Callback, CoroutineCallback
+from slixmpp.xmlstream.handler import CoroutineCallback
from slixmpp.xmlstream.matcher import StanzaPath
-from slixmpp.xmlstream import register_stanza_plugin, JID
+from slixmpp.xmlstream import register_stanza_plugin
from slixmpp.plugins.xep_0030 import stanza, DiscoInfo, DiscoItems
from slixmpp.plugins.xep_0030 import StaticDisco
@@ -42,14 +48,14 @@ class XEP_0030(BasePlugin):
Node handler hierarchy:
- ::
-
- JID | Node | Level
- ---------------------
- None | None | Global
- Given | None | All nodes for the JID
- None | Given | Node on self.xmpp.boundjid
- Given | Given | A single node
+ ====== ======= ============================
+ JID Node Level
+ ====== ======= ============================
+ None None Global
+ Given None All nodes for the JID
+ None Given Node on self.xmpp.boundjid
+ Given Given A single node
+ ====== ======= ============================
Stream Handlers:
@@ -62,19 +68,13 @@ class XEP_0030(BasePlugin):
Events:
- ::
-
- disco_info -- Received a disco#info Iq query result.
- disco_items -- Received a disco#items Iq query result.
- disco_info_query -- Received a disco#info Iq query request.
- disco_items_query -- Received a disco#items Iq query request.
+ - :term:`disco_info` -- Received a disco#info Iq query result.
+ - :term:`disco_items` -- Received a disco#items Iq query result.
Attributes:
:var static: Object containing the default set of
static node handlers.
- :var default_handlers: A dictionary mapping operations to the default
- global handler (by default, the static handlers).
"""
name = 'xep_0030'
@@ -85,20 +85,23 @@ class XEP_0030(BasePlugin):
'use_cache': True,
'wrap_results': False
}
+ static: StaticDisco
def plugin_init(self):
"""
Start the XEP-0030 plugin.
"""
- self.xmpp.register_handler(
- CoroutineCallback('Disco Info',
- StanzaPath('iq/disco_info'),
- self._handle_disco_info))
+ self.xmpp.register_handler(CoroutineCallback(
+ 'Disco Info',
+ StanzaPath('iq/disco_info'),
+ self._handle_disco_info
+ ))
- self.xmpp.register_handler(
- CoroutineCallback('Disco Items',
- StanzaPath('iq/disco_items'),
- self._handle_disco_items))
+ self.xmpp.register_handler(CoroutineCallback(
+ 'Disco Items',
+ StanzaPath('iq/disco_items'),
+ self._handle_disco_items
+ ))
register_stanza_plugin(Iq, DiscoInfo)
register_stanza_plugin(Iq, DiscoItems)
@@ -127,7 +130,7 @@ class XEP_0030(BasePlugin):
self.api.register(default_handler, op)
self.api.register_default(default_handler, op)
- def set_node_handler(self, htype: str, jid: Optional[JID] = None,
+ def set_node_handler(self, htype: str, jid: OptJid = None,
node: Optional[str] = None,
handler: Optional[Callable] = None):
"""
@@ -142,14 +145,14 @@ class XEP_0030(BasePlugin):
Node handler hierarchy:
- ::
-
- JID | Node | Level
- ---------------------
- None | None | Global
- Given | None | All nodes for the JID
- None | Given | Node on self.xmpp.boundjid
- Given | Given | A single node
+ ====== ======= ============================
+ JID Node Level
+ ====== ======= ============================
+ None None Global
+ Given None All nodes for the JID
+ None Given Node on self.xmpp.boundjid
+ Given Given A single node
+ ====== ======= ============================
Handler types:
@@ -180,7 +183,7 @@ class XEP_0030(BasePlugin):
"""
self.api.register(handler, htype, jid, node)
- def del_node_handler(self, htype, jid, node):
+ def del_node_handler(self, htype: str, jid: OptJid, node: Optional[str]):
"""
Remove a handler type for a JID and node combination.
@@ -190,14 +193,14 @@ class XEP_0030(BasePlugin):
Node handler hierarchy:
- ::
-
- JID | Node | Level
- ---------------------
- None | None | Global
- Given | None | All nodes for the JID
- None | Given | Node on self.xmpp.boundjid
- Given | Given | A single node
+ ====== ======= ============================
+ JID Node Level
+ ====== ======= ============================
+ None None Global
+ Given None All nodes for the JID
+ None Given Node on self.xmpp.boundjid
+ Given Given A single node
+ ====== ======= ============================
:param htype: The type of handler to remove.
:param jid: The JID from which to remove the handler.
@@ -205,7 +208,8 @@ class XEP_0030(BasePlugin):
"""
self.api.unregister(htype, jid, node)
- def restore_defaults(self, jid=None, node=None, handlers=None):
+ def restore_defaults(self, jid: OptJid = None, node: Optional[str] = None,
+ handlers: Optional[List[Callable]] = None):
"""
Change all or some of a node's handlers to the default
handlers. Useful for manually overriding the contents
@@ -227,19 +231,15 @@ class XEP_0030(BasePlugin):
for op in handlers:
self.api.restore_default(op, jid, node)
- def supports(self, jid=None, node=None, feature=None, local=False,
- cached=True, ifrom=None) -> Future:
+ def supports(self, jid: OptJid = None, node: Optional[str] = None,
+ feature: Optional[str] = None, local: bool = False,
+ cached: bool = True, ifrom: OptJid = None) -> Future:
"""
Check if a JID supports a given feature.
.. versionchanged:: 1.8.0
This function now returns a Future.
- Return values:
- :param True: The feature is supported
- :param False: The feature is not listed as supported
- :param None: Nothing could be found due to a timeout
-
:param jid: Request info from this JID.
:param node: The particular node to query.
:param feature: The name of the feature to check.
@@ -255,24 +255,27 @@ class XEP_0030(BasePlugin):
be useful. If set to false, then the cache will
be skipped, even if a result has already been
cached. Defaults to false.
+
+ :returns True: The feature is supported
+ :returns False: The feature is not listed as supported
+ :returns None: Nothing could be found due to a timeout
"""
data = {'feature': feature,
'local': local,
'cached': cached}
return self.api['supports'](jid, node, ifrom, data)
- def has_identity(self, jid=None, node=None, category=None, itype=None,
- lang=None, local=False, cached=True, ifrom=None) -> Future:
+ def has_identity(self, jid: OptJid = None, node: Optional[str] = None,
+ category: Optional[str] = None,
+ itype: Optional[str] = None, lang: Optional[str] = None,
+ local: bool = False, cached: bool = True,
+ ifrom: OptJid = None) -> Future:
"""
Check if a JID provides a given identity.
.. versionchanged:: 1.8.0
This function now returns a Future.
- Return values:
- :param True: The identity is provided
- :param False: The identity is not listed
- :param None: Nothing could be found due to a timeout
:param jid: Request info from this JID.
:param node: The particular node to query.
@@ -291,6 +294,10 @@ class XEP_0030(BasePlugin):
be useful. If set to false, then the cache will
be skipped, even if a result has already been
cached. Defaults to false.
+
+ :returns True: The identity is provided
+ :returns False: The identity is not listed
+ :returns None: Nothing could be found due to a timeout
"""
data = {'category': category,
'itype': itype,
@@ -301,7 +308,8 @@ class XEP_0030(BasePlugin):
async def get_info_from_domain(self, domain=None, timeout=None,
cached=True, callback=None):
- """Fetch disco#info of specified domain and one disco#items level below"""
+ """Fetch disco#info of specified domain and one disco#items level below
+ """
if domain is None:
domain = self.xmpp.boundjid.domain
@@ -322,7 +330,9 @@ class XEP_0030(BasePlugin):
)
self.domain_infos[domain] = [
- future.result() for future in info_futures if not future.exception()]
+ future.result()
+ for future in info_futures if not future.exception()
+ ]
results = self.domain_infos[domain]
@@ -330,8 +340,9 @@ class XEP_0030(BasePlugin):
callback(results)
return results
- async def get_info(self, jid=None, node=None, local=None,
- cached=None, **kwargs):
+ async def get_info(self, jid: OptJid = None, node: Optional[str] = None,
+ local: Optional[bool] = None,
+ cached: Optional[bool] = None, **kwargs) -> Iq:
"""
Retrieve the disco#info results from a given JID/node combination.
@@ -375,7 +386,7 @@ class XEP_0030(BasePlugin):
local = True
if local:
- log.debug("Looking up local disco#info data " + \
+ log.debug("Looking up local disco#info data "
"for %s, node %s.", jid, node)
info = await self.api['get_info'](
jid, node, kwargs.get('ifrom', None),
@@ -385,7 +396,7 @@ class XEP_0030(BasePlugin):
return self._wrap(kwargs.get('ifrom', None), jid, info)
if cached:
- log.debug("Looking up cached disco#info data " + \
+ log.debug("Looking up cached disco#info data "
"for %s, node %s.", jid, node)
info = await self.api['get_cached_info'](
jid, node,
@@ -401,11 +412,10 @@ class XEP_0030(BasePlugin):
iq['to'] = jid
iq['type'] = 'get'
iq['disco_info']['node'] = node if node else ''
- return await iq.send(timeout=kwargs.get('timeout', None),
- callback=kwargs.get('callback', None),
- timeout_callback=kwargs.get('timeout_callback', None))
+ return await iq.send(**kwargs)
- def set_info(self, jid=None, node=None, info=None) -> Future:
+ def set_info(self, jid: OptJid = None, node: Optional[str] = None,
+ info: Optional[Union[Iq, DiscoInfo]] = None) -> Future:
"""
Set the disco#info data for a JID/node based on an existing
disco#info stanza.
@@ -418,7 +428,9 @@ class XEP_0030(BasePlugin):
info = info['disco_info']
return self.api['set_info'](jid, node, None, info)
- async def get_items(self, jid=None, node=None, local=False, **kwargs):
+ async def get_items(self, jid: OptJid = None, node: Optional[str] = None,
+ local: bool = False, ifrom: OptJid = None,
+ **kwargs) -> Iq:
"""
Retrieve the disco#items results from a given JID/node combination.
@@ -445,25 +457,22 @@ class XEP_0030(BasePlugin):
Otherwise the parameter is ignored.
"""
if local or local is None and jid is None:
- items = await self.api['get_items'](jid, node,
- kwargs.get('ifrom', None),
- kwargs)
+ items = await self.api['get_items'](jid, node, ifrom, kwargs)
return self._wrap(kwargs.get('ifrom', None), jid, items)
iq = self.xmpp.Iq()
# Check dfrom parameter for backwards compatibility
- iq['from'] = kwargs.get('ifrom', kwargs.get('dfrom', ''))
+ iq['from'] = ifrom or kwargs.get('dfrom', '')
iq['to'] = jid
iq['type'] = 'get'
iq['disco_items']['node'] = node if node else ''
if kwargs.get('iterator', False) and self.xmpp['xep_0059']:
return self.xmpp['xep_0059'].iterate(iq, 'disco_items')
else:
- return await iq.send(timeout=kwargs.get('timeout', None),
- callback=kwargs.get('callback', None),
- timeout_callback=kwargs.get('timeout_callback', None))
+ return await iq.send(**kwargs)
- def set_items(self, jid=None, node=None, **kwargs) -> Future:
+ def set_items(self, jid: OptJid = None, node: Optional[str] = None,
+ **kwargs) -> Future:
"""
Set or replace all items for the specified JID/node combination.
@@ -480,7 +489,8 @@ class XEP_0030(BasePlugin):
"""
return self.api['set_items'](jid, node, None, kwargs)
- def del_items(self, jid=None, node=None, **kwargs) -> Future:
+ def del_items(self, jid: OptJid = None, node: Optional[str] = None,
+ **kwargs) -> Future:
"""
Remove all items from the given JID/node combination.
@@ -493,7 +503,9 @@ class XEP_0030(BasePlugin):
"""
return self.api['del_items'](jid, node, None, kwargs)
- def add_item(self, jid='', name='', node=None, subnode='', ijid=None) -> Future:
+ def add_item(self, jid: str = '', name: str = '',
+ node: Optional[str] = None, subnode: str = '',
+ ijid: OptJid = None) -> Future:
"""
Add a new item element to the given JID/node combination.
@@ -516,7 +528,8 @@ class XEP_0030(BasePlugin):
'inode': subnode}
return self.api['add_item'](ijid, node, None, kwargs)
- def del_item(self, jid=None, node=None, **kwargs) -> Future:
+ def del_item(self, jid: OptJid = None, node: Optional[str] = None,
+ **kwargs) -> Future:
"""
Remove a single item from the given JID/node combination.
@@ -527,8 +540,9 @@ class XEP_0030(BasePlugin):
"""
return self.api['del_item'](jid, node, None, kwargs)
- def add_identity(self, category='', itype='', name='',
- node=None, jid=None, lang=None) -> Future:
+ def add_identity(self, category: str = '', itype: str = '', name: str = '',
+ node: Optional[str] = None, jid: OptJid = None,
+ lang: Optional[str] = None) -> Future:
"""
Add a new identity to the given JID/node combination.
@@ -557,7 +571,7 @@ class XEP_0030(BasePlugin):
return self.api['add_identity'](jid, node, None, kwargs)
def add_feature(self, feature: str, node: Optional[str] = None,
- jid: Optional[JID] = None) -> Future:
+ jid: OptJid = None) -> Future:
"""
Add a feature to a JID/node combination.
@@ -571,7 +585,7 @@ class XEP_0030(BasePlugin):
kwargs = {'feature': feature}
return self.api['add_feature'](jid, node, None, kwargs)
- def del_identity(self, jid: Optional[JID] = None,
+ def del_identity(self, jid: OptJid = None,
node: Optional[str] = None, **kwargs) -> Future:
"""
Remove an identity from the given JID/node combination.
@@ -588,7 +602,8 @@ class XEP_0030(BasePlugin):
"""
return self.api['del_identity'](jid, node, None, kwargs)
- def del_feature(self, jid=None, node=None, **kwargs) -> Future:
+ def del_feature(self, jid: OptJid = None, node: Optional[str] = None,
+ **kwargs) -> Future:
"""
Remove a feature from a given JID/node combination.
@@ -601,7 +616,8 @@ class XEP_0030(BasePlugin):
"""
return self.api['del_feature'](jid, node, None, kwargs)
- def set_identities(self, jid=None, node=None, **kwargs) -> Future:
+ def set_identities(self, jid: OptJid = None, node: Optional[str] = None,
+ **kwargs) -> Future:
"""
Add or replace all identities for the given JID/node combination.
@@ -618,7 +634,8 @@ class XEP_0030(BasePlugin):
"""
return self.api['set_identities'](jid, node, None, kwargs)
- def del_identities(self, jid=None, node=None, **kwargs) -> Future:
+ def del_identities(self, jid: OptJid = None, node: Optional[str] = None,
+ **kwargs) -> Future:
"""
Remove all identities for a JID/node combination.
@@ -635,7 +652,8 @@ class XEP_0030(BasePlugin):
"""
return self.api['del_identities'](jid, node, None, kwargs)
- def set_features(self, jid=None, node=None, **kwargs) -> Future:
+ def set_features(self, jid: OptJid = None, node: Optional[str] = None,
+ **kwargs) -> Future:
"""
Add or replace the set of supported features
for a JID/node combination.
@@ -649,7 +667,8 @@ class XEP_0030(BasePlugin):
"""
return self.api['set_features'](jid, node, None, kwargs)
- def del_features(self, jid=None, node=None, **kwargs) -> Future:
+ def del_features(self, jid: OptJid = None, node: Optional[str] = None,
+ **kwargs) -> Future:
"""
Remove all features from a JID/node combination.
@@ -661,7 +680,8 @@ class XEP_0030(BasePlugin):
"""
return self.api['del_features'](jid, node, None, kwargs)
- async def _run_node_handler(self, htype, jid, node=None, ifrom=None, data=None):
+ async def _run_node_handler(self, htype, jid, node: Optional[str] = None,
+ ifrom: OptJid = None, data=None):
"""
Execute the most specific node handler for the given
JID/node combination.
@@ -676,7 +696,7 @@ class XEP_0030(BasePlugin):
return await self.api[htype](jid, node, ifrom, data)
- async def _handle_disco_info(self, iq):
+ async def _handle_disco_info(self, iq: Iq):
"""
Process an incoming disco#info stanza. If it is a get
request, find and return the appropriate identities
@@ -686,7 +706,7 @@ class XEP_0030(BasePlugin):
:param iq: The incoming disco#items stanza.
"""
if iq['type'] == 'get':
- log.debug("Received disco info query from " + \
+ log.debug("Received disco info query from "
"<%s> to <%s>.", iq['from'], iq['to'])
info = await self.api['get_info'](iq['to'],
iq['disco_info']['node'],
@@ -704,11 +724,11 @@ class XEP_0030(BasePlugin):
iq.set_payload(info.xml)
iq.send()
elif iq['type'] == 'result':
- log.debug("Received disco info result from " + \
+ log.debug("Received disco info result from "
"<%s> to <%s>.", iq['from'], iq['to'])
if self.use_cache:
- log.debug("Caching disco info result from " \
- "<%s> to <%s>.", iq['from'], iq['to'])
+ log.debug("Caching disco info result from "
+ "<%s> to <%s>.", iq['from'], iq['to'])
if self.xmpp.is_component:
ito = iq['to'].full
else:
@@ -719,7 +739,7 @@ class XEP_0030(BasePlugin):
iq)
self.xmpp.event('disco_info', iq)
- async def _handle_disco_items(self, iq):
+ async def _handle_disco_items(self, iq: Iq):
"""
Process an incoming disco#items stanza. If it is a get
request, find and return the appropriate items. If it
@@ -728,7 +748,7 @@ class XEP_0030(BasePlugin):
:param iq: The incoming disco#items stanza.
"""
if iq['type'] == 'get':
- log.debug("Received disco items query from " + \
+ log.debug("Received disco items query from "
"<%s> to <%s>.", iq['from'], iq['to'])
items = await self.api['get_items'](iq['to'],
iq['disco_items']['node'],
@@ -742,11 +762,11 @@ class XEP_0030(BasePlugin):
iq.set_payload(items.xml)
iq.send()
elif iq['type'] == 'result':
- log.debug("Received disco items result from " + \
+ log.debug("Received disco items result from "
"%s to %s.", iq['from'], iq['to'])
self.xmpp.event('disco_items', iq)
- def _fix_default_info(self, info):
+ def _fix_default_info(self, info: DiscoInfo):
"""
Disco#info results for a JID are required to include at least
one identity and feature. As a default, if no other identity is
@@ -762,20 +782,20 @@ class XEP_0030(BasePlugin):
if not info['node']:
if not info['identities']:
if self.xmpp.is_component:
- log.debug("No identity found for this entity. " + \
+ log.debug("No identity found for this entity. "
"Using default component identity.")
info.add_identity('component', 'generic')
else:
- log.debug("No identity found for this entity. " + \
+ log.debug("No identity found for this entity. "
"Using default client identity.")
info.add_identity('client', 'bot')
if not info['features']:
- log.debug("No features found for this entity. " + \
+ log.debug("No features found for this entity. "
"Using default disco#info feature.")
info.add_feature(info.namespace)
return result
- def _wrap(self, ito, ifrom, payload, force=False):
+ def _wrap(self, ito: OptJid, ifrom: OptJid, payload, force=False) -> Iq:
"""
Ensure that results are wrapped in an Iq stanza
if self.wrap_results has been set to True.
diff --git a/slixmpp/plugins/xep_0030/stanza/info.py b/slixmpp/plugins/xep_0030/stanza/info.py
index a77917e9..367afdad 100644
--- a/slixmpp/plugins/xep_0030/stanza/info.py
+++ b/slixmpp/plugins/xep_0030/stanza/info.py
@@ -1,10 +1,19 @@
-
# Slixmpp: The Slick XMPP Library
# Copyright (C) 2010 Nathanael C. Fritz, Lance J.T. Stout
# This file is part of Slixmpp.
# See the file LICENSE for copying permission.
+from typing import (
+ Iterable,
+ List,
+ Optional,
+ Set,
+ Tuple,
+ Union,
+)
from slixmpp.xmlstream import ElementBase, ET
+IdentityType = Tuple[str, str, Optional[str], Optional[str]]
+
class DiscoInfo(ElementBase):
@@ -18,21 +27,23 @@ class DiscoInfo(ElementBase):
category with a type of 'pc' to indicate the agent is a human operated
client with a GUI, or a category of 'gateway' with a type of 'aim' to
identify the agent as a gateway for the legacy AIM protocol. See
- <http://xmpp.org/registrar/disco-categories.html> for a full list of
+ `XMPP Registrar Disco Categories`_ for a full list of
accepted category and type combinations.
+ .. _XMPP Registrar Disco Categories: <http://xmpp.org/registrar/disco-categories.html>
+
Features are simply a set of the namespaces that identify the supported
features. For example, a client that supports service discovery will
- include the feature 'http://jabber.org/protocol/disco#info'.
+ include the feature ``http://jabber.org/protocol/disco#info``.
Since clients and components may operate in several roles at once, identity
and feature information may be grouped into "nodes". If one were to write
all of the identities and features used by a client, then node names would
be like section headings.
- Example disco#info stanzas:
+ Example disco#info stanza:
- ::
+ .. code-block:: xml
<iq type="get">
<query xmlns="http://jabber.org/protocol/disco#info" />
@@ -46,30 +57,26 @@ class DiscoInfo(ElementBase):
<feature var="urn:xmpp:ping" />
</query>
</iq>
-
- Stanza Interface:
- ::
-
- node -- The name of the node to either
- query or return info from.
- identities -- A set of 4-tuples, where each tuple contains
- the category, type, xml:lang, and name
- of an identity.
- features -- A set of namespaces for features.
-
"""
name = 'query'
namespace = 'http://jabber.org/protocol/disco#info'
plugin_attrib = 'disco_info'
+ #: Stanza interfaces:
+ #:
+ #: - ``node``: The name of the node to either query or return the info from
+ #: - ``identities``: A set of 4-tuples, where each tuple contains the
+ #: category, type, xml:lang and name of an identity
+ #: - ``features``: A set of namespaces for features
+ #:
interfaces = {'node', 'features', 'identities'}
lang_interfaces = {'identities'}
# Cache identities and features
- _identities = set()
- _features = set()
+ _identities: Set[Tuple[str, str, Optional[str]]]
+ _features: Set[str]
- def setup(self, xml=None):
+ def setup(self, xml: Optional[ET.ElementTree] = None):
"""
Populate the stanza object using an optional XML object.
@@ -84,7 +91,9 @@ class DiscoInfo(ElementBase):
self._identities = {id[0:3] for id in self['identities']}
self._features = self['features']
- def add_identity(self, category, itype, name=None, lang=None):
+ def add_identity(self, category: str, itype: str,
+ name: Optional[str] = None, lang: Optional[str] = None
+ ) -> bool:
"""
Add a new identity element. Each identity must be unique
in terms of all four identity components.
@@ -113,7 +122,8 @@ class DiscoInfo(ElementBase):
return True
return False
- def del_identity(self, category, itype, name=None, lang=None):
+ def del_identity(self, category: str, itype: str, name=None,
+ lang: Optional[str] = None) -> bool:
"""
Remove a given identity.
@@ -134,7 +144,8 @@ class DiscoInfo(ElementBase):
return True
return False
- def get_identities(self, lang=None, dedupe=True):
+ def get_identities(self, lang: Optional[str] = None, dedupe: bool = True
+ ) -> Iterable[IdentityType]:
"""
Return a set of all identities in tuple form as so:
@@ -147,6 +158,7 @@ class DiscoInfo(ElementBase):
:param dedupe: If True, de-duplicate identities, otherwise
return a list of all identities.
"""
+ identities: Union[List[IdentityType], Set[IdentityType]]
if dedupe:
identities = set()
else:
@@ -158,13 +170,14 @@ class DiscoInfo(ElementBase):
id_xml.attrib['type'],
id_xml.attrib.get('{%s}lang' % self.xml_ns, None),
id_xml.attrib.get('name', None))
- if dedupe:
+ if isinstance(identities, set):
identities.add(id)
else:
identities.append(id)
return identities
- def set_identities(self, identities, lang=None):
+ def set_identities(self, identities: Iterable[IdentityType],
+ lang: Optional[str] = None):
"""
Add or replace all identities. The identities must be a in set
where each identity is a tuple of the form:
@@ -187,7 +200,7 @@ class DiscoInfo(ElementBase):
category, itype, lang, name = identity
self.add_identity(category, itype, name, lang)
- def del_identities(self, lang=None):
+ def del_identities(self, lang: Optional[str] = None):
"""
Remove all identities. If a language was specified, only
remove identities using that language.
@@ -204,7 +217,7 @@ class DiscoInfo(ElementBase):
id_xml.attrib.get('{%s}lang' % self.xml_ns, None)))
self.xml.remove(id_xml)
- def add_feature(self, feature):
+ def add_feature(self, feature: str) -> bool:
"""
Add a single, new feature.
@@ -218,7 +231,7 @@ class DiscoInfo(ElementBase):
return True
return False
- def del_feature(self, feature):
+ def del_feature(self, feature: str) -> bool:
"""
Remove a single feature.
@@ -232,20 +245,21 @@ class DiscoInfo(ElementBase):
return True
return False
- def get_features(self, dedupe=True):
+ def get_features(self, dedupe: bool = True) -> Iterable[str]:
"""Return the set of all supported features."""
+ features: Union[List[str], Set[str]]
if dedupe:
features = set()
else:
features = []
for feature_xml in self.xml.findall('{%s}feature' % self.namespace):
- if dedupe:
+ if isinstance(features, set):
features.add(feature_xml.attrib['var'])
else:
features.append(feature_xml.attrib['var'])
return features
- def set_features(self, features):
+ def set_features(self, features: Iterable[str]):
"""
Add or replace the set of supported features.
diff --git a/slixmpp/plugins/xep_0030/stanza/items.py b/slixmpp/plugins/xep_0030/stanza/items.py
index d322804c..a72a8140 100644
--- a/slixmpp/plugins/xep_0030/stanza/items.py
+++ b/slixmpp/plugins/xep_0030/stanza/items.py
@@ -1,9 +1,34 @@
-
# Slixmpp: The Slick XMPP Library
# Copyright (C) 2010 Nathanael C. Fritz, Lance J.T. Stout
# This file is part of Slixmpp.
# See the file LICENSE for copying permission.
-from slixmpp.xmlstream import ElementBase, register_stanza_plugin
+from typing import (
+ Iterable,
+ Optional,
+ Set,
+ Tuple,
+)
+from slixmpp import JID
+from slixmpp.xmlstream import (
+ ElementBase,
+ ET,
+ register_stanza_plugin,
+)
+
+
+class DiscoItem(ElementBase):
+ name = 'item'
+ namespace = 'http://jabber.org/protocol/disco#items'
+ plugin_attrib = name
+ interfaces = {'jid', 'node', 'name'}
+
+ def get_node(self) -> Optional[str]:
+ """Return the item's node name or ``None``."""
+ return self._get_attr('node', None)
+
+ def get_name(self) -> Optional[str]:
+ """Return the item's human readable name, or ``None``."""
+ return self._get_attr('name', None)
class DiscoItems(ElementBase):
@@ -11,7 +36,7 @@ class DiscoItems(ElementBase):
"""
Example disco#items stanzas:
- ::
+ .. code-block:: xml
<iq type="get">
<query xmlns="http://jabber.org/protocol/disco#items" />
@@ -28,25 +53,24 @@ class DiscoItems(ElementBase):
</query>
</iq>
- Stanza Interface:
- ::
-
- node -- The name of the node to either
- query or return info from.
- items -- A list of 3-tuples, where each tuple contains
- the JID, node, and name of an item.
-
"""
name = 'query'
namespace = 'http://jabber.org/protocol/disco#items'
plugin_attrib = 'disco_items'
+ #: Stanza Interface:
+ #:
+ #: - ``node``: The name of the node to either
+ #: query or return info from.
+ #: - ``items``: A list of 3-tuples, where each tuple contains
+ #: the JID, node, and name of an item.
+ #:
interfaces = {'node', 'items'}
# Cache items
- _items = set()
+ _items: Set[Tuple[JID, Optional[str]]]
- def setup(self, xml=None):
+ def setup(self, xml: Optional[ET.ElementTree] = None):
"""
Populate the stanza object using an optional XML object.
@@ -59,7 +83,8 @@ class DiscoItems(ElementBase):
ElementBase.setup(self, xml)
self._items = {item[0:2] for item in self['items']}
- def add_item(self, jid, node=None, name=None):
+ def add_item(self, jid: JID, node: Optional[str] = None,
+ name: Optional[str] = None):
"""
Add a new item element. Each item is required to have a
JID, but may also specify a node value to reference
@@ -80,7 +105,7 @@ class DiscoItems(ElementBase):
return True
return False
- def del_item(self, jid, node=None):
+ def del_item(self, jid: JID, node: Optional[str] = None) -> bool:
"""
Remove a single item.
@@ -96,7 +121,7 @@ class DiscoItems(ElementBase):
return True
return False
- def get_items(self):
+ def get_items(self) -> Set[DiscoItem]:
"""Return all items."""
items = set()
for item in self['substanzas']:
@@ -104,7 +129,7 @@ class DiscoItems(ElementBase):
items.add((item['jid'], item['node'], item['name']))
return items
- def set_items(self, items):
+ def set_items(self, items: Iterable[DiscoItem]):
"""
Set or replace all items. The given items must be in a
list or set where each item is a tuple of the form:
@@ -127,19 +152,4 @@ class DiscoItems(ElementBase):
self.iterables.remove(item)
-class DiscoItem(ElementBase):
- name = 'item'
- namespace = 'http://jabber.org/protocol/disco#items'
- plugin_attrib = name
- interfaces = {'jid', 'node', 'name'}
-
- def get_node(self):
- """Return the item's node name or ``None``."""
- return self._get_attr('node', None)
-
- def get_name(self):
- """Return the item's human readable name, or ``None``."""
- return self._get_attr('name', None)
-
-
register_stanza_plugin(DiscoItems, DiscoItem, iterable=True)
diff --git a/slixmpp/plugins/xep_0030/static.py b/slixmpp/plugins/xep_0030/static.py
index 1ae34148..6863830f 100644
--- a/slixmpp/plugins/xep_0030/static.py
+++ b/slixmpp/plugins/xep_0030/static.py
@@ -1,20 +1,46 @@
-
# Slixmpp: The Slick XMPP Library
# Copyright (C) 2010 Nathanael C. Fritz, Lance J.T. Stout
# This file is part of Slixmpp.
# See the file LICENSE for copying permission.
+from __future__ import annotations
+
import logging
-from slixmpp import Iq
+from typing import (
+ Optional,
+ Any,
+ Dict,
+ Tuple,
+ TYPE_CHECKING,
+ Union,
+ Collection,
+)
+
+from slixmpp import BaseXMPP, JID
+from slixmpp.stanza import Iq
+from slixmpp.types import TypedDict, OptJidStr, OptJid
from slixmpp.exceptions import XMPPError, IqError, IqTimeout
-from slixmpp.xmlstream import JID
from slixmpp.plugins.xep_0030 import DiscoInfo, DiscoItems
log = logging.getLogger(__name__)
+if TYPE_CHECKING:
+ from slixmpp.plugins.xep_0030 import XEP_0030
+
+
+class NodeType(TypedDict):
+ info: DiscoInfo
+ items: DiscoItems
+
+
+NodesType = Dict[
+ Tuple[str, str, str],
+ NodeType
+]
-class StaticDisco(object):
+
+class StaticDisco:
"""
While components will likely require fully dynamic handling
@@ -25,75 +51,87 @@ class StaticDisco(object):
StaticDisco provides a set of node handlers that will store
static sets of disco info and items in memory.
- Attributes:
- nodes -- A dictionary mapping (JID, node) tuples to a dict
- containing a disco#info and a disco#items stanza.
- xmpp -- The main Slixmpp object.
+ :var nodes: A dictionary mapping (JID, node) tuples to a dict
+ containing a disco#info and a disco#items stanza.
+ :var xmpp: The main Slixmpp object.
+ :var disco: The instance of the XEP-0030 plugin.
"""
- def __init__(self, xmpp, disco):
+ def __init__(self, xmpp: 'BaseXMPP', disco: 'XEP_0030'):
"""
Create a static disco interface. Sets of disco#info and
disco#items are maintained for every given JID and node
combination. These stanzas are used to store disco
information in memory without any additional processing.
- Arguments:
- xmpp -- The main Slixmpp object.
- """
- self.nodes = {}
- self.xmpp = xmpp
- self.disco = disco
-
- def add_node(self, jid=None, node=None, ifrom=None):
+ :param xmpp: The main Slixmpp object.
+ :param disco: The XEP-0030 plugin.
"""
- Create a new set of stanzas for the provided
- JID and node combination.
+ self.nodes: NodesType = {}
+ self.xmpp: BaseXMPP = xmpp
+ self.disco: 'XEP_0030' = disco
- Arguments:
- jid -- The JID that will own the new stanzas.
- node -- The node that will own the new stanzas.
- """
+ def add_node(self, jid: OptJidStr = None, node: Optional[str] = None,
+ ifrom: OptJidStr = None) -> NodeType:
if jid is None:
- jid = self.xmpp.boundjid.full
+ node_jid = self.xmpp.boundjid.full
+ elif isinstance(jid, JID):
+ node_jid = jid.full
+ if ifrom is None:
+ node_ifrom = ''
+ elif isinstance(ifrom, JID):
+ node_ifrom = ifrom.full
+ else:
+ node_ifrom = ifrom
if node is None:
node = ''
- if ifrom is None:
- ifrom = ''
- if isinstance(ifrom, JID):
- ifrom = ifrom.full
- if (jid, node, ifrom) not in self.nodes:
- new_node = {'info': DiscoInfo(), 'items': DiscoItems()}
- new_node['info']['node'] = node
- new_node['items']['node'] = node
- self.nodes[(jid, node, ifrom)] = new_node
- return self.nodes[(jid, node, ifrom)]
-
- def get_node(self, jid=None, node=None, ifrom=None):
+ if (node_jid, node, node_ifrom) not in self.nodes:
+ node_dict: NodeType = {
+ 'info': DiscoInfo(),
+ 'items': DiscoItems(),
+ }
+ node_dict['info']['node'] = node
+ node_dict['items']['node'] = node
+ self.nodes[(node_jid, node, node_ifrom)] = node_dict
+ return self.nodes[(node_jid, node, node_ifrom)]
+
+ def get_node(self, jid: OptJidStr = None, node: Optional[str] = None,
+ ifrom: OptJidStr = None) -> NodeType:
if jid is None:
- jid = self.xmpp.boundjid.full
+ node_jid = self.xmpp.boundjid.full
+ elif isinstance(jid, JID):
+ node_jid = jid.full
+ else:
+ node_jid = jid
if node is None:
node = ''
if ifrom is None:
- ifrom = ''
- if isinstance(ifrom, JID):
- ifrom = ifrom.full
- if (jid, node, ifrom) not in self.nodes:
- self.add_node(jid, node, ifrom)
- return self.nodes[(jid, node, ifrom)]
-
- def node_exists(self, jid=None, node=None, ifrom=None):
+ node_ifrom = ''
+ elif isinstance(ifrom, JID):
+ node_ifrom = ifrom.full
+ else:
+ node_ifrom = ifrom
+ if (node_jid, node, node_ifrom) not in self.nodes:
+ self.add_node(node_jid, node, node_ifrom)
+ return self.nodes[(node_jid, node, node_ifrom)]
+
+ def node_exists(self, jid: OptJidStr = None, node: Optional[str] = None,
+ ifrom: OptJidStr = None) -> bool:
if jid is None:
- jid = self.xmpp.boundjid.full
+ node_jid = self.xmpp.boundjid.full
+ elif isinstance(jid, JID):
+ node_jid = jid.full
+ else:
+ node_jid = jid
if node is None:
node = ''
if ifrom is None:
- ifrom = ''
- if isinstance(ifrom, JID):
- ifrom = ifrom.full
- if (jid, node, ifrom) not in self.nodes:
- return False
- return True
+ node_ifrom = ''
+ elif isinstance(ifrom, JID):
+ node_ifrom = ifrom.full
+ else:
+ node_ifrom = ifrom
+ return (node_jid, node, node_ifrom) in self.nodes
# =================================================================
# Node Handlers
@@ -109,18 +147,20 @@ class StaticDisco(object):
# the requester's JID, except for cached results. To do that,
# register a custom node handler.
- async def supports(self, jid, node, ifrom, data):
+ async def supports(self, jid: OptJid, node: Optional[str], ifrom: OptJid,
+ data: Any) -> Optional[bool]:
"""
Check if a JID supports a given feature.
The data parameter may provide:
- feature -- The feature to check for support.
- local -- If true, then the query is for a JID/node
+
+ :param feature: The feature to check for support.
+ :param local: If true, then the query is for a JID/node
combination handled by this Slixmpp instance and
no stanzas need to be sent.
Otherwise, a disco stanza must be sent to the
remove JID to retrieve the info.
- cached -- If true, then look for the disco info data from
+ :param cached: If true, then look for the disco info data from
the local cache system. If no results are found,
send the query as usual. The self.use_cache
setting must be set to true for this option to
@@ -147,26 +187,29 @@ class StaticDisco(object):
except IqTimeout:
return None
- async def has_identity(self, jid, node, ifrom, data):
+ async def has_identity(self, jid: OptJid, node: Optional[str],
+ ifrom: OptJid, data: Dict[str, Any]
+ ) -> Optional[bool]:
"""
Check if a JID has a given identity.
The data parameter may provide:
- category -- The category of the identity to check.
- itype -- The type of the identity to check.
- lang -- The language of the identity to check.
- local -- If true, then the query is for a JID/node
- combination handled by this Slixmpp instance and
- no stanzas need to be sent.
- Otherwise, a disco stanza must be sent to the
- remove JID to retrieve the info.
- cached -- If true, then look for the disco info data from
- the local cache system. If no results are found,
- send the query as usual. The self.use_cache
- setting must be set to true for this option to
- be useful. If set to false, then the cache will
- be skipped, even if a result has already been
- cached. Defaults to false.
+
+ :param category: The category of the identity to check.
+ :param itype: The type of the identity to check.
+ :param lang: The language of the identity to check.
+ :param local: If true, then the query is for a JID/node
+ combination handled by this Slixmpp instance and
+ no stanzas need to be sent.
+ Otherwise, a disco stanza must be sent to the
+ remove JID to retrieve the info.
+ :param cached: If true, then look for the disco info data from
+ the local cache system. If no results are found,
+ send the query as usual. The self.use_cache
+ setting must be set to true for this option to
+ be useful. If set to false, then the cache will
+ be skipped, even if a result has already been
+ cached. Defaults to false.
"""
identity = (data.get('category', None),
data.get('itype', None),
@@ -179,14 +222,17 @@ class StaticDisco(object):
info = await self.disco.get_info(jid=jid, node=node,
ifrom=ifrom, **data)
info = self.disco._wrap(ifrom, jid, info, True)
- trunc = lambda i: (i[0], i[1], i[2])
+
+ def trunc(i):
+ return (i[0], i[1], i[2])
return identity in map(trunc, info['disco_info']['identities'])
except IqError:
return False
except IqTimeout:
return None
- def get_info(self, jid, node, ifrom, data):
+ def get_info(self, jid: OptJid, node: Optional[str], ifrom: OptJid,
+ data: Any) -> Optional[DiscoInfo]:
"""
Return the stored info data for the requested JID/node combination.
@@ -200,7 +246,8 @@ class StaticDisco(object):
else:
return self.get_node(jid, node)['info']
- def set_info(self, jid, node, ifrom, data):
+ def set_info(self, jid: OptJid, node: Optional[str], ifrom: OptJid,
+ data: DiscoInfo):
"""
Set the entire info stanza for a JID/node at once.
@@ -209,7 +256,8 @@ class StaticDisco(object):
new_node = self.add_node(jid, node)
new_node['info'] = data
- def del_info(self, jid, node, ifrom, data):
+ def del_info(self, jid: OptJid, node: Optional[str], ifrom: OptJid,
+ data: Any):
"""
Reset the info stanza for a given JID/node combination.
@@ -218,7 +266,8 @@ class StaticDisco(object):
if self.node_exists(jid, node):
self.get_node(jid, node)['info'] = DiscoInfo()
- def get_items(self, jid, node, ifrom, data):
+ def get_items(self, jid: OptJid, node: Optional[str], ifrom: OptJid,
+ data: Any) -> Optional[DiscoItems]:
"""
Return the stored items data for the requested JID/node combination.
@@ -232,7 +281,8 @@ class StaticDisco(object):
else:
return self.get_node(jid, node)['items']
- def set_items(self, jid, node, ifrom, data):
+ def set_items(self, jid: OptJid, node: Optional[str], ifrom: OptJid,
+ data: Dict[str, Collection[Tuple]]):
"""
Replace the stored items data for a JID/node combination.
@@ -243,7 +293,8 @@ class StaticDisco(object):
new_node = self.add_node(jid, node)
new_node['items']['items'] = items
- def del_items(self, jid, node, ifrom, data):
+ def del_items(self, jid: OptJid, node: Optional[str], ifrom: OptJid,
+ data: Any):
"""
Reset the items stanza for a given JID/node combination.
@@ -252,15 +303,17 @@ class StaticDisco(object):
if self.node_exists(jid, node):
self.get_node(jid, node)['items'] = DiscoItems()
- def add_identity(self, jid, node, ifrom, data):
+ def add_identity(self, jid: OptJid, node: Optional[str], ifrom: OptJid,
+ data: Dict[str, Optional[str]]):
"""
Add a new identity to the JID/node combination.
The data parameter may provide:
- category -- The general category to which the agent belongs.
- itype -- A more specific designation with the category.
- name -- Optional human readable name for this identity.
- lang -- Optional standard xml:lang value.
+
+ :param category: The general category to which the agent belongs.
+ :param itype: A more specific designation with the category.
+ :param name: Optional human readable name for this identity.
+ :param lang: Optional standard xml:lang value.
"""
new_node = self.add_node(jid, node)
new_node['info'].add_identity(
@@ -269,27 +322,31 @@ class StaticDisco(object):
data.get('name', None),
data.get('lang', None))
- def set_identities(self, jid, node, ifrom, data):
+ def set_identities(self, jid: OptJid, node: Optional[str], ifrom: OptJid,
+ data: Dict[str, Collection[str]]):
"""
Add or replace all identities for a JID/node combination.
The data parameter should include:
- identities -- A list of identities in tuple form:
- (category, type, name, lang)
+
+ :param identities: A list of identities in tuple form:
+ (category, type, name, lang)
"""
identities = data.get('identities', set())
new_node = self.add_node(jid, node)
new_node['info']['identities'] = identities
- def del_identity(self, jid, node, ifrom, data):
+ def del_identity(self, jid: OptJid, node: Optional[str], ifrom: OptJid,
+ data: Dict[str, Optional[str]]):
"""
Remove an identity from a JID/node combination.
The data parameter may provide:
- category -- The general category to which the agent belonged.
- itype -- A more specific designation with the category.
- name -- Optional human readable name for this identity.
- lang -- Optional, standard xml:lang value.
+
+ :param category: The general category to which the agent belonged.
+ :param itype: A more specific designation with the category.
+ :param name: Optional human readable name for this identity.
+ :param lang: Optional, standard xml:lang value.
"""
if self.node_exists(jid, node):
self.get_node(jid, node)['info'].del_identity(
@@ -298,7 +355,8 @@ class StaticDisco(object):
data.get('name', None),
data.get('lang', None))
- def del_identities(self, jid, node, ifrom, data):
+ def del_identities(self, jid: OptJid, node: Optional[str], ifrom: OptJid,
+ data: Any):
"""
Remove all identities from a JID/node combination.
@@ -307,40 +365,47 @@ class StaticDisco(object):
if self.node_exists(jid, node):
del self.get_node(jid, node)['info']['identities']
- def add_feature(self, jid, node, ifrom, data):
+ def add_feature(self, jid: OptJid, node: Optional[str], ifrom: OptJid,
+ data: Dict[str, str]):
"""
Add a feature to a JID/node combination.
The data parameter should include:
- feature -- The namespace of the supported feature.
+
+ :param feature: The namespace of the supported feature.
"""
new_node = self.add_node(jid, node)
new_node['info'].add_feature(
data.get('feature', ''))
- def set_features(self, jid, node, ifrom, data):
+ def set_features(self, jid: OptJid, node: Optional[str], ifrom: OptJid,
+ data: Dict[str, Collection[str]]):
"""
Add or replace all features for a JID/node combination.
The data parameter should include:
- features -- The new set of supported features.
+
+ :param features: The new set of supported features.
"""
features = data.get('features', set())
new_node = self.add_node(jid, node)
new_node['info']['features'] = features
- def del_feature(self, jid, node, ifrom, data):
+ def del_feature(self, jid: OptJid, node: Optional[str], ifrom: OptJid,
+ data: Dict[str, str]):
"""
Remove a feature from a JID/node combination.
The data parameter should include:
- feature -- The namespace of the removed feature.
+
+ :param feature: The namespace of the removed feature.
"""
if self.node_exists(jid, node):
self.get_node(jid, node)['info'].del_feature(
data.get('feature', ''))
- def del_features(self, jid, node, ifrom, data):
+ def del_features(self, jid: OptJid, node: Optional[str], ifrom: OptJid,
+ data: Any):
"""
Remove all features from a JID/node combination.
@@ -350,15 +415,17 @@ class StaticDisco(object):
return
del self.get_node(jid, node)['info']['features']
- def add_item(self, jid, node, ifrom, data):
+ def add_item(self, jid: OptJid, node: Optional[str], ifrom: OptJid,
+ data: Dict[str, str]):
"""
Add an item to a JID/node combination.
The data parameter may include:
- ijid -- The JID for the item.
- inode -- Optional additional information to reference
- non-addressable items.
- name -- Optional human readable name for the item.
+
+ :param ijid: The JID for the item.
+ :param inode: Optional additional information to reference
+ non-addressable items.
+ :param name: Optional human readable name for the item.
"""
new_node = self.add_node(jid, node)
new_node['items'].add_item(
@@ -366,20 +433,23 @@ class StaticDisco(object):
node=data.get('inode', ''),
name=data.get('name', ''))
- def del_item(self, jid, node, ifrom, data):
+ def del_item(self, jid: OptJid, node: Optional[str], ifrom: OptJid,
+ data: Dict[str, str]):
"""
Remove an item from a JID/node combination.
The data parameter may include:
- ijid -- JID of the item to remove.
- inode -- Optional extra identifying information.
+
+ :param ijid: JID of the item to remove.
+ :param inode: Optional extra identifying information.
"""
if self.node_exists(jid, node):
self.get_node(jid, node)['items'].del_item(
data.get('ijid', ''),
node=data.get('inode', None))
- def cache_info(self, jid, node, ifrom, data):
+ def cache_info(self, jid: OptJid, node: Optional[str], ifrom: OptJid,
+ data: Union[Iq, DiscoInfo]):
"""
Cache disco information for an external JID.
@@ -388,12 +458,15 @@ class StaticDisco(object):
the disco#info substanza itself.
"""
if isinstance(data, Iq):
- data = data['disco_info']
+ info = data['disco_info']
+ else:
+ info = data
new_node = self.add_node(jid, node, ifrom)
- new_node['info'] = data
+ new_node['info'] = info
- def get_cached_info(self, jid, node, ifrom, data):
+ def get_cached_info(self, jid: OptJid, node: Optional[str], ifrom: OptJid,
+ data: Any) -> Optional[DiscoInfo]:
"""
Retrieve cached disco info data.
@@ -401,5 +474,4 @@ class StaticDisco(object):
"""
if not self.node_exists(jid, node, ifrom):
return None
- else:
- return self.get_node(jid, node, ifrom)['info']
+ return self.get_node(jid, node, ifrom)['info']
diff --git a/slixmpp/types.py b/slixmpp/types.py
index c8ab640c..b9a1d319 100644
--- a/slixmpp/types.py
+++ b/slixmpp/types.py
@@ -7,7 +7,10 @@
This file contains boilerplate to define types relevant to slixmpp.
"""
-from typing import Optional
+from typing import (
+ Optional,
+ Union,
+)
try:
from typing import (
@@ -31,7 +34,6 @@ PresenceShows = Literal[
'away', 'chat', 'dnd', 'xa',
]
-
MessageTypes = Literal[
'chat', 'error', 'groupchat',
'headline', 'normal',
@@ -70,3 +72,7 @@ class MucRoomItem(TypedDict, total=False):
MucRoomItemKeys = Literal[
'jid', 'role', 'affiliation', 'show', 'status', 'alt_nick',
]
+
+OptJid = Optional[JID]
+JidStr = Union[str, JID]
+OptJidStr = Optional[Union[str, JID]]