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

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

Browse Source
This commit is contained in:
mathieui 2021-02-04 21:28:18 +01:00
parent 99c2e5cafd
commit 712ac671e1
1 changed files with 14 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

33
slixmpp/plugins/xep_0152/reachability.py
View File

@ -8,6 +8,8 @@
import logging import logging
from asyncio import Future
from slixmpp import JID from slixmpp import JID
from typing import Dict, List, Optional, Callable from typing import Dict, List, Optional, Callable
from slixmpp.plugins.base import BasePlugin from slixmpp.plugins.base import BasePlugin
@ -37,17 +39,12 @@ class XEP_0152(BasePlugin):
self.xmpp['xep_0163'].register_pep('reachability', Reachability) self.xmpp['xep_0163'].register_pep('reachability', Reachability)
def publish_reachability(self, addresses: List[Dict[str, str]], def publish_reachability(self, addresses: List[Dict[str, str]],
options: Optional[Form] = None, **pubsubkwargs) -> Future:
ifrom: Optional[JID] = None,
callback: Optional[Callable] = None,
timeout: Optional[int] = None,
timeout_callback: Optional[Callable] = None):
""" """
Publish alternative addresses where the user can be reached. Publish alternative addresses where the user can be reached.
:param addresses: A list of dictionaries containing the URI and :param addresses: A list of dictionaries containing the URI and
optional description for each address. optional description for each address.
:param options: Optional form of publish options.
""" """
if not isinstance(addresses, (list, tuple)): if not isinstance(addresses, (list, tuple)):
addresses = [addresses] addresses = [addresses]
@ -60,25 +57,19 @@ class XEP_0152(BasePlugin):
for key, val in address.items(): for key, val in address.items():
addr[key] = val addr[key] = val
reach.append(addr) reach.append(addr)
return self.xmpp['xep_0163'].publish(reach, return self.xmpp['xep_0163'].publish(
reach,
node=Reachability.namespace, node=Reachability.namespace,
options=options, **pubsubkwargs
ifrom=ifrom, )
callback=callback,
timeout=timeout,
timeout_callback=timeout_callback)
def stop(self, ifrom: Optional[JID] = None, def stop(self, **pubsubkwargs) -> Future:
callback: Optional[Callable] = None,
timeout: Optional[int] = None,
timeout_callback: Optional[Callable] = None):
""" """
Clear existing user activity information to stop notifications. Clear existing user activity information to stop notifications.
""" """
reach = Reachability() reach = Reachability()
return self.xmpp['xep_0163'].publish(reach, return self.xmpp['xep_0163'].publish(
reach,
node=Reachability.namespace, node=Reachability.namespace,
ifrom=ifrom, **pubsubkwargs
callback=callback, )
timeout=timeout,
timeout_callback=timeout_callback)