summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authormathieui <mathieui@mathieui.net>2021-03-01 20:55:46 +0100
committermathieui <mathieui@mathieui.net>2021-03-01 21:11:33 +0100
commit49416dacbcf3c4070c16a29370f76dfef6dfad46 (patch)
tree63c503719a62dd44952860d9247cc28b278a7769
parente3027dabb27e06680ef15b3cf740560a9f6ac937 (diff)
downloadslixmpp-49416dacbcf3c4070c16a29370f76dfef6dfad46.tar.gz
slixmpp-49416dacbcf3c4070c16a29370f76dfef6dfad46.tar.bz2
slixmpp-49416dacbcf3c4070c16a29370f76dfef6dfad46.tar.xz
slixmpp-49416dacbcf3c4070c16a29370f76dfef6dfad46.zip
XEP-0030 static: Fix PEP8, types
-rw-r--r--slixmpp/plugins/xep_0030/static.py300
1 files changed, 186 insertions, 114 deletions
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']