jshiffer/slixmpp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

XEP-0191: Add type hints and switch to default args

Browse Source
This commit is contained in:
mathieui 2021-02-04 22:05:50 +01:00
parent 69e04d7d2e
commit cb8d2edc8d
1 changed files with 34 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

57
slixmpp/plugins/xep_0191/blocking.py
View File

@ -8,7 +8,15 @@
import logging import logging
from slixmpp import Iq from asyncio import Future
from typing import (
List,
Optional,
Set,
Union,
)
from slixmpp.stanza import Iq
from slixmpp.plugins import BasePlugin from slixmpp.plugins import BasePlugin
from slixmpp.xmlstream.handler import Callback from slixmpp.xmlstream.handler import Callback
from slixmpp.xmlstream.matcher import StanzaPath from slixmpp.xmlstream.matcher import StanzaPath
@ -18,6 +26,12 @@ from slixmpp.plugins.xep_0191 import stanza, Block, Unblock, BlockList
log = logging.getLogger(__name__) log = logging.getLogger(__name__)
BlockedJIDs = Union[
JID,
Set[JID],
List[JID]
]
class XEP_0191(BasePlugin): class XEP_0191(BasePlugin):
@ -45,42 +59,39 @@ class XEP_0191(BasePlugin):
self.xmpp.remove_handler('Blocked Contact') self.xmpp.remove_handler('Blocked Contact')
self.xmpp.remove_handler('Unblocked Contact') self.xmpp.remove_handler('Unblocked Contact')
def get_blocked(self, ifrom=None, timeout=None, callback=None, def get_blocked(self, ifrom: Optional[JID] = None, **iqkwargs) -> Future:
timeout_callback=None): """Get the list of blocked JIDs."""
iq = self.xmpp.Iq() iq = self.xmpp.make_iq_get(ifrom=ifrom)
iq['type'] = 'get'
iq['from'] = ifrom
iq.enable('blocklist') iq.enable('blocklist')
return iq.send(timeout=timeout, callback=callback, return iq.send(**iqkwargs)
timeout_callback=timeout_callback)
def block(self, jids, ifrom=None, timeout=None, callback=None, def block(self, jids: BlockedJIDs,
timeout_callback=None): ifrom: Optional[JID] = None, **iqkwargs) -> Future:
iq = self.xmpp.Iq() """Block a JID or a list of JIDs.
iq['type'] = 'set'
iq['from'] = ifrom
:param jids: JID(s) to block.
"""
iq = self.xmpp.make_iq_set(ifrom=ifrom)
if not isinstance(jids, (set, list)): if not isinstance(jids, (set, list)):
jids = [jids] jids = [jids]
iq['block']['items'] = jids iq['block']['items'] = jids
return iq.send(timeout=timeout, callback=callback, return iq.send(**iqkwargs)
timeout_callback=timeout_callback)
def unblock(self, jids=None, ifrom=None, timeout=None, callback=None, def unblock(self, jids: BlockedJIDs, ifrom: Optional[JID] = None, **iqkwargs) -> Future:
timeout_callback=None): """Unblock a JID or a list of JIDs.
iq = self.xmpp.Iq()
iq['type'] = 'set'
iq['from'] = ifrom
:param jids: JID(s) to unblock.
"""
if jids is None: if jids is None:
jids = [] raise ValueError("jids cannot be empty.")
iq = self.xmpp.make_iq_set(ifrom=ifrom)
if not isinstance(jids, (set, list)): if not isinstance(jids, (set, list)):
jids = [jids] jids = [jids]
iq['unblock']['items'] = jids iq['unblock']['items'] = jids
return iq.send(timeout=timeout, callback=callback, return iq.send(**iqkwargs)
timeout_callback=timeout_callback)
def _handle_blocked(self, iq): def _handle_blocked(self, iq):
self.xmpp.event('blocked', iq) self.xmpp.event('blocked', iq)