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

basexmpp/clientxmpp: refine typing

Browse source
This commit is contained in:
mathieui 2021-07-03 11:12:28 +02:00
parent 7fc3c48cad
commit df963542c1
2 changed files with 58 additions and 60 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

74
slixmpp/basexmpp.py
View file

@ -45,10 +45,11 @@ log = logging.getLogger(__name__)
from slixmpp.types import (
PresenceShows,
PresenceTypes,
MessageTypes,
IqTypes,
JidStr,
OptJidStr,
)
if TYPE_CHECKING:
@ -314,8 +315,8 @@ class BaseXMPP(XMLStream):
pres['lang'] = self.default_lang
return pres
def make_iq(self, id: str = "0", ifrom: Optional[JID] = None,
ito: Optional[JID] = None, itype: Optional[IqTypes] = None,
def make_iq(self, id: str = "0", ifrom: OptJidStr = None,
ito: OptJidStr = None, itype: Optional[IqTypes] = None,
iquery: Optional[str] = None) -> Iq:
"""Create a new :class:`~.Iq` stanza with a given Id and from JID.
@ -339,7 +340,7 @@ class BaseXMPP(XMLStream):
return iq
def make_iq_get(self, queryxmlns: Optional[str] =None,
ito: Optional[JID] = None, ifrom: Optional[JID] = None,
ito: OptJidStr = None, ifrom: OptJidStr = None,
iq: Optional[Iq] = None) -> Iq:
"""Create an :class:`~.Iq` stanza of type ``'get'``.
@ -364,7 +365,7 @@ class BaseXMPP(XMLStream):
return iq
def make_iq_result(self, id: Optional[str] = None,
ito: Optional[JID] = None, ifrom: Optional[JID] = None,
ito: OptJidStr = None, ifrom: OptJidStr = None,
iq: Optional[Iq] = None) -> Iq:
"""
Create an :class:`~.Iq` stanza of type
@ -391,7 +392,7 @@ class BaseXMPP(XMLStream):
return iq
def make_iq_set(self, sub: Optional[Union[ElementBase, ET.Element]] = None,
ito: Optional[JID] = None, ifrom: Optional[JID] = None,
ito: OptJidStr = None, ifrom: OptJidStr = None,
iq: Optional[Iq] = None) -> Iq:
"""
Create an :class:`~.Iq` stanza of type ``'set'``.
@ -454,8 +455,8 @@ class BaseXMPP(XMLStream):
return iq
def make_iq_query(self, iq: Optional[Iq] = None, xmlns: str = '',
ito: Optional[JID] = None,
ifrom: Optional[JID] = None) -> Iq:
ito: OptJidStr = None,
ifrom: OptJidStr = None) -> Iq:
"""
Create or modify an :class:`~.Iq` stanza
to use the given query namespace.
@ -487,10 +488,10 @@ class BaseXMPP(XMLStream):
iq['query'] = 'jabber:iq:roster'
return ET.Element("{jabber:iq:roster}query")
def make_message(self, mto: JID, mbody: Optional[str] = None,
def make_message(self, mto: JidStr, mbody: Optional[str] = None,
msubject: Optional[str] = None,
mtype: Optional[MessageTypes] = None,
mhtml: Optional[str] = None, mfrom: Optional[JID] = None,
mhtml: Optional[str] = None, mfrom: OptJidStr = None,
mnick: Optional[str] = None) -> Message:
"""
Create and initialize a new
@ -516,12 +517,12 @@ class BaseXMPP(XMLStream):
message['html']['body'] = mhtml
return message
def make_presence(self, pshow: Optional[PresenceShows] = None,
def make_presence(self, pshow: Optional[str] = None,
pstatus: Optional[str] = None,
ppriority: Optional[int] = None,
pto: Optional[JID] = None,
pto: OptJidStr = None,
ptype: Optional[PresenceTypes] = None,
pfrom: Optional[JID] = None,
pfrom: OptJidStr = None,
pnick: Optional[str] = None) -> Presence:
"""
Create and initialize a new
@ -548,7 +549,7 @@ class BaseXMPP(XMLStream):
def send_message(self, mto: JID, mbody: Optional[str] = None,
msubject: Optional[str] = None,
mtype: Optional[MessageTypes] = None,
mhtml: Optional[str] = None, mfrom: Optional[JID] = None,
mhtml: Optional[str] = None, mfrom: OptJidStr = None,
mnick: Optional[str] = None):
"""
Create, initialize, and send a new
@ -568,12 +569,12 @@ class BaseXMPP(XMLStream):
self.make_message(mto, mbody, msubject, mtype,
mhtml, mfrom, mnick).send()
def send_presence(self, pshow: Optional[PresenceShows] = None,
def send_presence(self, pshow: Optional[str] = None,
pstatus: Optional[str] = None,
ppriority: Optional[int] = None,
pto: Optional[JID] = None,
pto: OptJidStr = None,
ptype: Optional[PresenceTypes] = None,
pfrom: Optional[JID] = None,
pfrom: OptJidStr = None,
pnick: Optional[str] = None):
"""
Create, initialize, and send a new
@ -590,8 +591,9 @@ class BaseXMPP(XMLStream):
self.make_presence(pshow, pstatus, ppriority, pto,
ptype, pfrom, pnick).send()
def send_presence_subscription(self, pto, pfrom=None,
ptype='subscribe', pnick=None):
def send_presence_subscription(self, pto: JidStr, pfrom: OptJidStr = None,
ptype: PresenceTypes='subscribe', pnick:
Optional[str] = None):
"""
Create, initialize, and send a new
:class:`~.Presence` stanza of
@ -608,62 +610,62 @@ class BaseXMPP(XMLStream):
pnick=pnick).send()
@property
def jid(self):
def jid(self) -> str:
"""Attribute accessor for bare jid"""
log.warning("jid property deprecated. Use boundjid.bare")
return self.boundjid.bare
@jid.setter
def jid(self, value):
def jid(self, value: str):
log.warning("jid property deprecated. Use boundjid.bare")
self.boundjid.bare = value
@property
def fulljid(self):
def fulljid(self) -> str:
"""Attribute accessor for full jid"""
log.warning("fulljid property deprecated. Use boundjid.full")
return self.boundjid.full
@fulljid.setter
def fulljid(self, value):
def fulljid(self, value: str):
log.warning("fulljid property deprecated. Use boundjid.full")
self.boundjid.full = value
@property
def resource(self):
def resource(self) -> str:
"""Attribute accessor for jid resource"""
log.warning("resource property deprecated. Use boundjid.resource")
return self.boundjid.resource
@resource.setter
def resource(self, value):
def resource(self, value: str):
log.warning("fulljid property deprecated. Use boundjid.resource")
self.boundjid.resource = value
@property
def username(self):
def username(self) -> str:
"""Attribute accessor for jid usernode"""
log.warning("username property deprecated. Use boundjid.user")
return self.boundjid.user
@username.setter
def username(self, value):
def username(self, value: str):
log.warning("username property deprecated. Use boundjid.user")
self.boundjid.user = value
@property
def server(self):
def server(self) -> str:
"""Attribute accessor for jid host"""
log.warning("server property deprecated. Use boundjid.host")
return self.boundjid.server
@server.setter
def server(self, value):
def server(self, value: str):
log.warning("server property deprecated. Use boundjid.host")
self.boundjid.server = value
@property
def auto_authorize(self):
def auto_authorize(self) -> Optional[bool]:
"""Auto accept or deny subscription requests.
If ``True``, auto accept subscription requests.
@ -673,11 +675,11 @@ class BaseXMPP(XMLStream):
return self.roster.auto_authorize
@auto_authorize.setter
def auto_authorize(self, value):
def auto_authorize(self, value: Optional[bool]):
self.roster.auto_authorize = value
@property
def auto_subscribe(self):
def auto_subscribe(self) -> bool:
"""Auto send requests for mutual subscriptions.
If ``True``, auto send mutual subscription requests.
@ -685,21 +687,21 @@ class BaseXMPP(XMLStream):
return self.roster.auto_subscribe
@auto_subscribe.setter
def auto_subscribe(self, value):
def auto_subscribe(self, value: bool):
self.roster.auto_subscribe = value
def set_jid(self, jid):
def set_jid(self, jid: JidStr):
"""Rip a JID apart and claim it as our own."""
log.debug("setting jid to %s", jid)
self.boundjid = JID(jid)
def getjidresource(self, fulljid):
def getjidresource(self, fulljid: str):
if '/' in fulljid:
return fulljid.split('/', 1)[-1]
else:
return ''
def getjidbare(self, fulljid):
def getjidbare(self, fulljid: str):
return fulljid.split('/', 1)[0]
def _handle_session_start(self, event):

44
slixmpp/clientxmpp.py
View file

@ -8,23 +8,17 @@
# :license: MIT, see LICENSE for more details
import asyncio
import logging
from typing import Optional, Any, Callable, Tuple
from slixmpp.jid import JID
from slixmpp.stanza import StreamFeatures
from slixmpp.stanza import StreamFeatures, Iq
from slixmpp.basexmpp import BaseXMPP
from slixmpp.exceptions import XMPPError
from slixmpp.types import JidStr
from slixmpp.xmlstream import XMLStream
from slixmpp.xmlstream.matcher import StanzaPath, MatchXPath
from slixmpp.xmlstream.handler import Callback, CoroutineCallback
# Flag indicating if DNS SRV records are available for use.
try:
import dns.resolver
except ImportError:
DNSPYTHON = False
else:
DNSPYTHON = True
log = logging.getLogger(__name__)
@ -53,7 +47,7 @@ class ClientXMPP(BaseXMPP):
:param escape_quotes: **Deprecated.**
"""
def __init__(self, jid, password, plugin_config=None,
def __init__(self, jid: JidStr, password: str, plugin_config=None,
plugin_whitelist=None, escape_quotes=True, sasl_mech=None,
lang='en', **kwargs):
if not plugin_whitelist:
@ -103,7 +97,8 @@ class ClientXMPP(BaseXMPP):
CoroutineCallback('Stream Features',
MatchXPath('{%s}features' % self.stream_ns),
self._handle_stream_features))
def roster_push_filter(iq):
def roster_push_filter(iq: Iq) -> None:
from_ = iq['from']
if from_ and from_ != JID('') and from_ != self.boundjid.bare:
reply = iq.reply()
@ -131,15 +126,15 @@ class ClientXMPP(BaseXMPP):
self['feature_mechanisms'].use_mech = sasl_mech
@property
def password(self):
def password(self) -> str:
return self.credentials.get('password', '')
@password.setter
def password(self, value):
def password(self, value: str) -> None:
self.credentials['password'] = value
def connect(self, address=tuple(), use_ssl=False,
force_starttls=True, disable_starttls=False):
def connect(self, address: Optional[Tuple[str, int]] = None, use_ssl: bool = False,
force_starttls: bool = True, disable_starttls: bool = False) -> None:
"""Connect to the XMPP server.
When no address is given, a SRV lookup for the server will
@ -161,14 +156,15 @@ class ClientXMPP(BaseXMPP):
# XMPP client port and allow SRV lookup.
if address:
self.dns_service = None
host, port = address
else:
address = (self.boundjid.host, 5222)
host, port = (self.boundjid.host, 5222)
self.dns_service = 'xmpp-client'
return XMLStream.connect(self, address[0], address[1], use_ssl=use_ssl,
return XMLStream.connect(self, host, port, use_ssl=use_ssl,
force_starttls=force_starttls, disable_starttls=disable_starttls)
def register_feature(self, name, handler, restart=False, order=5000):
def register_feature(self, name: str, handler: Callable, restart: bool = False, order: int = 5000) -> None:
"""Register a stream feature handler.
:param name: The name of the stream feature.
@ -183,13 +179,13 @@ class ClientXMPP(BaseXMPP):
self._stream_feature_order.append((order, name))
self._stream_feature_order.sort()
def unregister_feature(self, name, order):
def unregister_feature(self, name: str, order: int) -> None:
if name in self._stream_feature_handlers:
del self._stream_feature_handlers[name]
self._stream_feature_order.remove((order, name))
self._stream_feature_order.sort()
def update_roster(self, jid, **kwargs):
def update_roster(self, jid: JID, **kwargs) -> None:
"""Add or change a roster item.
:param jid: The JID of the entry to modify.
@ -251,7 +247,7 @@ class ClientXMPP(BaseXMPP):
return iq.send(callback, timeout, timeout_callback)
def _reset_connection_state(self, event=None):
def _reset_connection_state(self, event: Optional[Any] = None) -> None:
#TODO: Use stream state here
self.authenticated = False
self.sessionstarted = False
@ -259,7 +255,7 @@ class ClientXMPP(BaseXMPP):
self.bindfail = False
self.features = set()
async def _handle_stream_features(self, features):
async def _handle_stream_features(self, features: StreamFeatures) -> Optional[bool]:
"""Process the received stream features.
:param features: The features stanza.
@ -278,7 +274,7 @@ class ClientXMPP(BaseXMPP):
log.debug('Finished processing stream features.')
self.event('stream_negotiated')
def _handle_roster(self, iq):
def _handle_roster(self, iq: Iq) -> None:
"""Update the roster after receiving a roster stanza.
:param iq: The roster stanza.
@ -310,7 +306,7 @@ class ClientXMPP(BaseXMPP):
resp.enable('roster')
resp.send()
def _handle_session_bind(self, jid):
def _handle_session_bind(self, jid: JID) -> None:
"""Set the client roster to the JID set by the server.
:param :class:`slixmpp.xmlstream.jid.JID` jid: The bound JID as