pep/slixmpp
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

basexmpp: add more typing, fix some docs

Browse source
This commit is contained in:
mathieui 2021-02-06 12:29:14 +01:00
parent d850b9a9f7
commit 622cfd4ed7
1 changed files with 92 additions and 47 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

139
slixmpp/basexmpp.py
View file

@ -11,6 +11,9 @@ import asyncio
import logging import logging
from typing import ( from typing import (
Dict,
Optional,
Union,
TYPE_CHECKING, TYPE_CHECKING,
) )
@ -18,14 +21,22 @@ from slixmpp import plugins, roster, stanza
from slixmpp.api import APIRegistry from slixmpp.api import APIRegistry
from slixmpp.exceptions import IqError, IqTimeout from slixmpp.exceptions import IqError, IqTimeout
from slixmpp.stanza import Message, Presence, Iq, StreamError from slixmpp.stanza import (
Message,
Presence,
Iq,
StreamError,
)
from slixmpp.stanza.roster import Roster from slixmpp.stanza.roster import Roster
from slixmpp.xmlstream import XMLStream, JID from slixmpp.xmlstream import XMLStream, JID
from slixmpp.xmlstream import ET, register_stanza_plugin from slixmpp.xmlstream import ET, register_stanza_plugin
from slixmpp.xmlstream.matcher import MatchXPath from slixmpp.xmlstream.matcher import MatchXPath
from slixmpp.xmlstream.handler import Callback from slixmpp.xmlstream.handler import Callback
from slixmpp.xmlstream.stanzabase import XML_NS from slixmpp.xmlstream.stanzabase import (
ElementBase,
XML_NS,
)
from slixmpp.plugins import PluginManager, load_plugin from slixmpp.plugins import PluginManager, load_plugin
@ -33,8 +44,16 @@ from slixmpp.plugins import PluginManager, load_plugin
log = logging.getLogger(__name__) log = logging.getLogger(__name__)
from slixmpp.types import (
PresenceShows,
PresenceTypes,
MessageTypes,
IqTypes,
)
if TYPE_CHECKING: if TYPE_CHECKING:
from slixmpp.types import PluginsDict # Circular imports
from slixmpp.pluginsdict import PluginsDict
class BaseXMPP(XMLStream): class BaseXMPP(XMLStream):
@ -229,7 +248,7 @@ class BaseXMPP(XMLStream):
self.plugin[name].post_init() self.plugin[name].post_init()
self.plugin[name].post_inited = True self.plugin[name].post_inited = True
def register_plugin(self, plugin, pconfig=None, module=None): def register_plugin(self, plugin: str, pconfig: Optional[Dict] = None, module=None):
"""Register and configure a plugin for use in this stream. """Register and configure a plugin for use in this stream.
:param plugin: The name of the plugin class. Plugin names must :param plugin: The name of the plugin class. Plugin names must
@ -279,32 +298,34 @@ class BaseXMPP(XMLStream):
"""Return a plugin given its name, if it has been registered.""" """Return a plugin given its name, if it has been registered."""
return self.plugin.get(key, default) return self.plugin.get(key, default)
def Message(self, *args, **kwargs): def Message(self, *args, **kwargs) -> Message:
"""Create a Message stanza associated with this stream.""" """Create a Message stanza associated with this stream."""
msg = Message(self, *args, **kwargs) msg = Message(self, *args, **kwargs)
msg['lang'] = self.default_lang msg['lang'] = self.default_lang
return msg return msg
def Iq(self, *args, **kwargs): def Iq(self, *args, **kwargs) -> Iq:
"""Create an Iq stanza associated with this stream.""" """Create an Iq stanza associated with this stream."""
return Iq(self, *args, **kwargs) return Iq(self, *args, **kwargs)
def Presence(self, *args, **kwargs): def Presence(self, *args, **kwargs) -> Presence:
"""Create a Presence stanza associated with this stream.""" """Create a Presence stanza associated with this stream."""
pres = Presence(self, *args, **kwargs) pres = Presence(self, *args, **kwargs)
pres['lang'] = self.default_lang pres['lang'] = self.default_lang
return pres return pres
def make_iq(self, id=0, ifrom=None, ito=None, itype=None, iquery=None): def make_iq(self, id: str = "0", ifrom: Optional[JID] = None,
"""Create a new Iq stanza with a given Id and from JID. ito: Optional[JID] = None, itype: Optional[IqTypes] = None,
iquery: Optional[str] = None) -> Iq:
"""Create a new :class:`~.Iq` stanza with a given Id and from JID.
:param id: An ideally unique ID value for this stanza thread. :param id: An ideally unique ID value for this stanza thread.
Defaults to 0. Defaults to 0.
:param ifrom: The from :class:`~slixmpp.xmlstream.jid.JID` :param ifrom: The from :class:`~.JID`
to use for this stanza. to use for this stanza.
:param ito: The destination :class:`~slixmpp.xmlstream.jid.JID` :param ito: The destination :class:`~.JID`
for this stanza. for this stanza.
:param itype: The :class:`~slixmpp.stanza.iq.Iq`'s type, :param itype: The :class:`~.Iq`'s type,
one of: ``'get'``, ``'set'``, ``'result'``, one of: ``'get'``, ``'set'``, ``'result'``,
or ``'error'``. or ``'error'``.
:param iquery: Optional namespace for adding a query element. :param iquery: Optional namespace for adding a query element.
@ -317,15 +338,17 @@ class BaseXMPP(XMLStream):
iq['query'] = iquery iq['query'] = iquery
return iq return iq
def make_iq_get(self, queryxmlns=None, ito=None, ifrom=None, iq=None): def make_iq_get(self, queryxmlns: Optional[str] =None,
"""Create an :class:`~slixmpp.stanza.iq.Iq` stanza of type ``'get'``. ito: Optional[JID] = None, ifrom: Optional[JID] = None,
iq: Optional[Iq] = None) -> Iq:
"""Create an :class:`~.Iq` stanza of type ``'get'``.
Optionally, a query element may be added. Optionally, a query element may be added.
:param queryxmlns: The namespace of the query to use. :param queryxmlns: The namespace of the query to use.
:param ito: The destination :class:`~slixmpp.xmlstream.jid.JID` :param ito: The destination :class:`~.JID`
for this stanza. for this stanza.
:param ifrom: The ``'from'`` :class:`~slixmpp.xmlstream.jid.JID` :param ifrom: The ``'from'`` :class:`~.JID`
to use for this stanza. to use for this stanza.
:param iq: Optionally use an existing stanza instead :param iq: Optionally use an existing stanza instead
of generating a new one. of generating a new one.
@ -340,15 +363,17 @@ class BaseXMPP(XMLStream):
iq['from'] = ifrom iq['from'] = ifrom
return iq return iq
def make_iq_result(self, id=None, ito=None, ifrom=None, iq=None): def make_iq_result(self, id: Optional[str] = None,
ito: Optional[JID] = None, ifrom: Optional[JID] = None,
iq: Optional[Iq] = None) -> Iq:
""" """
Create an :class:`~slixmpp.stanza.iq.Iq` stanza of type Create an :class:`~.Iq` stanza of type
``'result'`` with the given ID value. ``'result'`` with the given ID value.
:param id: An ideally unique ID value. May use :meth:`new_id()`. :param id: An ideally unique ID value. May use :meth:`new_id()`.
:param ito: The destination :class:`~slixmpp.xmlstream.jid.JID` :param ito: The destination :class:`~.JID`
for this stanza. for this stanza.
:param ifrom: The ``'from'`` :class:`~slixmpp.xmlstream.jid.JID` :param ifrom: The ``'from'`` :class:`~.JID`
to use for this stanza. to use for this stanza.
:param iq: Optionally use an existing stanza instead :param iq: Optionally use an existing stanza instead
of generating a new one. of generating a new one.
@ -365,21 +390,23 @@ class BaseXMPP(XMLStream):
iq['from'] = ifrom iq['from'] = ifrom
return iq return iq
def make_iq_set(self, sub=None, ito=None, ifrom=None, iq=None): def make_iq_set(self, sub: Optional[Union[ElementBase, ET.Element]] = None,
ito: Optional[JID] = None, ifrom: Optional[JID] = None,
iq: Optional[Iq] = None) -> Iq:
""" """
Create an :class:`~slixmpp.stanza.iq.Iq` stanza of type ``'set'``. Create an :class:`~.Iq` stanza of type ``'set'``.
Optionally, a substanza may be given to use as the Optionally, a substanza may be given to use as the
stanza's payload. stanza's payload.
:param sub: Either an :param sub: Either an
:class:`~slixmpp.xmlstream.stanzabase.ElementBase` :class:`~.ElementBase`
stanza object or an stanza object or an
:class:`~xml.etree.ElementTree.Element` XML object :class:`~xml.etree.ElementTree.Element` XML object
to use as the :class:`~slixmpp.stanza.iq.Iq`'s payload. to use as the :class:`~.Iq`'s payload.
:param ito: The destination :class:`~slixmpp.xmlstream.jid.JID` :param ito: The destination :class:`~.JID`
for this stanza. for this stanza.
:param ifrom: The ``'from'`` :class:`~slixmpp.xmlstream.jid.JID` :param ifrom: The ``'from'`` :class:`~.JID`
to use for this stanza. to use for this stanza.
:param iq: Optionally use an existing stanza instead :param iq: Optionally use an existing stanza instead
of generating a new one. of generating a new one.
@ -399,7 +426,7 @@ class BaseXMPP(XMLStream):
condition='feature-not-implemented', condition='feature-not-implemented',
text=None, ito=None, ifrom=None, iq=None): text=None, ito=None, ifrom=None, iq=None):
""" """
Create an :class:`~slixmpp.stanza.iq.Iq` stanza of type ``'error'``. Create an :class:`~.Iq` stanza of type ``'error'``.
:param id: An ideally unique ID value. May use :meth:`new_id()`. :param id: An ideally unique ID value. May use :meth:`new_id()`.
:param type: The type of the error, such as ``'cancel'`` or :param type: The type of the error, such as ``'cancel'`` or
@ -407,9 +434,9 @@ class BaseXMPP(XMLStream):
:param condition: The error condition. Defaults to :param condition: The error condition. Defaults to
``'feature-not-implemented'``. ``'feature-not-implemented'``.
:param text: A message describing the cause of the error. :param text: A message describing the cause of the error.
:param ito: The destination :class:`~slixmpp.xmlstream.jid.JID` :param ito: The destination :class:`~.JID`
for this stanza. for this stanza.
:param ifrom: The ``'from'`` :class:`~slixmpp.xmlstream.jid.JID` :param ifrom: The ``'from'`` :class:`~jid.JID`
to use for this stanza. to use for this stanza.
:param iq: Optionally use an existing stanza instead :param iq: Optionally use an existing stanza instead
of generating a new one. of generating a new one.
@ -426,17 +453,19 @@ class BaseXMPP(XMLStream):
iq['from'] = ifrom iq['from'] = ifrom
return iq return iq
def make_iq_query(self, iq=None, xmlns='', ito=None, ifrom=None): def make_iq_query(self, iq: Optional[Iq] = None, xmlns: str = '',
ito: Optional[JID] = None,
ifrom: Optional[JID] = None) -> Iq:
""" """
Create or modify an :class:`~slixmpp.stanza.iq.Iq` stanza Create or modify an :class:`~.Iq` stanza
to use the given query namespace. to use the given query namespace.
:param iq: Optionally use an existing stanza instead :param iq: Optionally use an existing stanza instead
of generating a new one. of generating a new one.
:param xmlns: The query's namespace. :param xmlns: The query's namespace.
:param ito: The destination :class:`~slixmpp.xmlstream.jid.JID` :param ito: The destination :class:`~.JID`
for this stanza. for this stanza.
:param ifrom: The ``'from'`` :class:`~slixmpp.xmlstream.jid.JID` :param ifrom: The ``'from'`` :class:`~.JID`
to use for this stanza. to use for this stanza.
""" """
if not iq: if not iq:
@ -448,7 +477,7 @@ class BaseXMPP(XMLStream):
iq['from'] = ifrom iq['from'] = ifrom
return iq return iq
def make_query_roster(self, iq=None): def make_query_roster(self, iq: Optional[Iq] = None) -> ET.Element:
"""Create a roster query element. """Create a roster query element.
:param iq: Optionally use an existing stanza instead :param iq: Optionally use an existing stanza instead
@ -458,11 +487,14 @@ class BaseXMPP(XMLStream):
iq['query'] = 'jabber:iq:roster' iq['query'] = 'jabber:iq:roster'
return ET.Element("{jabber:iq:roster}query") return ET.Element("{jabber:iq:roster}query")
def make_message(self, mto, mbody=None, msubject=None, mtype=None, def make_message(self, mto: JID, mbody: Optional[str] = None,
mhtml=None, mfrom=None, mnick=None): msubject: Optional[str] = None,
mtype: Optional[MessageTypes] = None,
mhtml: Optional[str] = None, mfrom: Optional[JID] = None,
mnick: Optional[str] = None) -> Message:
""" """
Create and initialize a new Create and initialize a new
:class:`~slixmpp.stanza.message.Message` stanza. :class:`~.Message` stanza.
:param mto: The recipient of the message. :param mto: The recipient of the message.
:param mbody: The main contents of the message. :param mbody: The main contents of the message.
@ -484,11 +516,16 @@ class BaseXMPP(XMLStream):
message['html']['body'] = mhtml message['html']['body'] = mhtml
return message return message
def make_presence(self, pshow=None, pstatus=None, ppriority=None, def make_presence(self, pshow: Optional[PresenceShows] = None,
pto=None, ptype=None, pfrom=None, pnick=None): pstatus: Optional[str] = None,
ppriority: Optional[int] = None,
pto: Optional[JID] = None,
ptype: Optional[PresenceTypes] = None,
pfrom: Optional[JID] = None,
pnick: Optional[str] = None) -> Presence:
""" """
Create and initialize a new Create and initialize a new
:class:`~slixmpp.stanza.presence.Presence` stanza. :class:`~.Presence` stanza.
:param pshow: The presence's show value. :param pshow: The presence's show value.
:param pstatus: The presence's status message. :param pstatus: The presence's status message.
@ -508,11 +545,14 @@ class BaseXMPP(XMLStream):
presence['nick'] = pnick presence['nick'] = pnick
return presence return presence
def send_message(self, mto, mbody, msubject=None, mtype=None, def send_message(self, mto: JID, mbody: Optional[str] = None,
mhtml=None, mfrom=None, mnick=None): msubject: Optional[str] = None,
mtype: Optional[MessageTypes] = None,
mhtml: Optional[str] = None, mfrom: Optional[JID] = None,
mnick: Optional[str] = None):
""" """
Create, initialize, and send a new Create, initialize, and send a new
:class:`~slixmpp.stanza.message.Message` stanza. :class:`~.Message` stanza.
:param mto: The recipient of the message. :param mto: The recipient of the message.
:param mbody: The main contents of the message. :param mbody: The main contents of the message.
@ -528,11 +568,16 @@ class BaseXMPP(XMLStream):
self.make_message(mto, mbody, msubject, mtype, self.make_message(mto, mbody, msubject, mtype,
mhtml, mfrom, mnick).send() mhtml, mfrom, mnick).send()
def send_presence(self, pshow=None, pstatus=None, ppriority=None, def send_presence(self, pshow: Optional[PresenceShows] = None,
pto=None, pfrom=None, ptype=None, pnick=None): pstatus: Optional[str] = None,
ppriority: Optional[int] = None,
pto: Optional[JID] = None,
ptype: Optional[PresenceTypes] = None,
pfrom: Optional[JID] = None,
pnick: Optional[str] = None):
""" """
Create, initialize, and send a new Create, initialize, and send a new
:class:`~slixmpp.stanza.presence.Presence` stanza. :class:`~.Presence` stanza.
:param pshow: The presence's show value. :param pshow: The presence's show value.
:param pstatus: The presence's status message. :param pstatus: The presence's status message.
@ -549,7 +594,7 @@ class BaseXMPP(XMLStream):
ptype='subscribe', pnick=None): ptype='subscribe', pnick=None):
""" """
Create, initialize, and send a new Create, initialize, and send a new
:class:`~slixmpp.stanza.presence.Presence` stanza of :class:`~.Presence` stanza of
type ``'subscribe'``. type ``'subscribe'``.
:param pto: The recipient of a directed presence. :param pto: The recipient of a directed presence.