From fed55d3dda2c01dca7e9b9ea036c4b7b756510ff Mon Sep 17 00:00:00 2001 From: mathieui Date: Wed, 21 Apr 2021 23:20:25 +0200 Subject: typing: matchers and senders Leftover error that I cannot fix: * https://github.com/python/mypy/issues/708 Leftover error that I am unsure of what to do: * xml handlers are not properly typed (it seems like nothing in slix is using it, considering a removal instead of adding an Union everywhere) --- slixmpp/xmlstream/handler/base.py | 31 ++++++++++++----- slixmpp/xmlstream/handler/callback.py | 28 +++++++++++----- slixmpp/xmlstream/handler/collector.py | 40 ++++++++++++---------- slixmpp/xmlstream/handler/coroutine_callback.py | 38 ++++++++++++++------- slixmpp/xmlstream/handler/waiter.py | 44 +++++++++++++++++-------- slixmpp/xmlstream/handler/xmlcallback.py | 3 +- slixmpp/xmlstream/handler/xmlwaiter.py | 3 +- slixmpp/xmlstream/matcher/base.py | 9 +++-- slixmpp/xmlstream/matcher/id.py | 8 +++-- slixmpp/xmlstream/matcher/idsender.py | 15 +++++++-- slixmpp/xmlstream/matcher/many.py | 5 ++- slixmpp/xmlstream/matcher/stanzapath.py | 25 +++++++++----- slixmpp/xmlstream/matcher/xmlmask.py | 28 +++++++--------- slixmpp/xmlstream/matcher/xpath.py | 17 +++++----- 14 files changed, 189 insertions(+), 105 deletions(-) (limited to 'slixmpp') diff --git a/slixmpp/xmlstream/handler/base.py b/slixmpp/xmlstream/handler/base.py index 1e657777..e1edb249 100644 --- a/slixmpp/xmlstream/handler/base.py +++ b/slixmpp/xmlstream/handler/base.py @@ -4,10 +4,18 @@ # Part of Slixmpp: The Slick XMPP Library # :copyright: (c) 2011 Nathanael C. Fritz # :license: MIT, see LICENSE for more details +from __future__ import annotations + import weakref +from weakref import ReferenceType +from typing import Optional, TYPE_CHECKING +from slixmpp.xmlstream.matcher.base import MatcherBase + +if TYPE_CHECKING: + from slixmpp.xmlstream import XMLStream, StanzaBase -class BaseHandler(object): +class BaseHandler: """ Base class for stream handlers. Stream handlers are matched with @@ -26,8 +34,13 @@ class BaseHandler(object): :param stream: The :class:`~slixmpp.xmlstream.xmlstream.XMLStream` instance that the handle will respond to. """ + name: str + stream: Optional[ReferenceType[XMLStream]] + _destroy: bool + _matcher: MatcherBase + _payload: Optional[StanzaBase] - def __init__(self, name, matcher, stream=None): + def __init__(self, name: str, matcher: MatcherBase, stream: Optional[XMLStream] = None): #: The name of the handler self.name = name @@ -41,33 +54,33 @@ class BaseHandler(object): self._payload = None self._matcher = matcher - def match(self, xml): + def match(self, xml: StanzaBase) -> bool: """Compare a stanza or XML object with the handler's matcher. :param xml: An XML or - :class:`~slixmpp.xmlstream.stanzabase.ElementBase` object + :class:`~slixmpp.xmlstream.stanzabase.StanzaBase` object """ return self._matcher.match(xml) - def prerun(self, payload): + def prerun(self, payload: StanzaBase) -> None: """Prepare the handler for execution while the XML stream is being processed. - :param payload: A :class:`~slixmpp.xmlstream.stanzabase.ElementBase` + :param payload: A :class:`~slixmpp.xmlstream.stanzabase.StanzaBase` object. """ self._payload = payload - def run(self, payload): + def run(self, payload: StanzaBase) -> None: """Execute the handler after XML stream processing and during the main event loop. - :param payload: A :class:`~slixmpp.xmlstream.stanzabase.ElementBase` + :param payload: A :class:`~slixmpp.xmlstream.stanzabase.StanzaBase` object. """ self._payload = payload - def check_delete(self): + def check_delete(self) -> bool: """Check if the handler should be removed from the list of stream handlers. """ diff --git a/slixmpp/xmlstream/handler/callback.py b/slixmpp/xmlstream/handler/callback.py index 93cec6b7..5d792bf9 100644 --- a/slixmpp/xmlstream/handler/callback.py +++ b/slixmpp/xmlstream/handler/callback.py @@ -1,10 +1,17 @@ - # slixmpp.xmlstream.handler.callback # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ # Part of Slixmpp: The Slick XMPP Library # :copyright: (c) 2011 Nathanael C. Fritz # :license: MIT, see LICENSE for more details +from __future__ import annotations + +from typing import Optional, Callable, Any, TYPE_CHECKING from slixmpp.xmlstream.handler.base import BaseHandler +from slixmpp.xmlstream.matcher.base import MatcherBase + +if TYPE_CHECKING: + from slixmpp.xmlstream.stanzabase import StanzaBase + from slixmpp.xmlstream.xmlstream import XMLStream class Callback(BaseHandler): @@ -28,8 +35,6 @@ class Callback(BaseHandler): :param matcher: A :class:`~slixmpp.xmlstream.matcher.base.MatcherBase` derived object for matching stanza objects. :param pointer: The function to execute during callback. - :param bool thread: **DEPRECATED.** Remains only for - backwards compatibility. :param bool once: Indicates if the handler should be used only once. Defaults to False. :param bool instream: Indicates if the callback should be executed @@ -38,31 +43,36 @@ class Callback(BaseHandler): :param stream: The :class:`~slixmpp.xmlstream.xmlstream.XMLStream` instance this handler should monitor. """ + _once: bool + _instream: bool + _pointer: Callable[[StanzaBase], Any] - def __init__(self, name, matcher, pointer, thread=False, - once=False, instream=False, stream=None): + def __init__(self, name: str, matcher: MatcherBase, + pointer: Callable[[StanzaBase], Any], + once: bool = False, instream: bool = False, + stream: Optional[XMLStream] = None): BaseHandler.__init__(self, name, matcher, stream) self._pointer = pointer self._once = once self._instream = instream - def prerun(self, payload): + def prerun(self, payload: StanzaBase) -> None: """Execute the callback during stream processing, if the callback was created with ``instream=True``. :param payload: The matched - :class:`~slixmpp.xmlstream.stanzabase.ElementBase` object. + :class:`~slixmpp.xmlstream.stanzabase.StanzaBase` object. """ if self._once: self._destroy = True if self._instream: self.run(payload, True) - def run(self, payload, instream=False): + def run(self, payload: StanzaBase, instream: bool = False) -> None: """Execute the callback function with the matched stanza payload. :param payload: The matched - :class:`~slixmpp.xmlstream.stanzabase.ElementBase` object. + :class:`~slixmpp.xmlstream.stanzabase.StanzaBase` object. :param bool instream: Force the handler to execute during stream processing. This should only be used by :meth:`prerun()`. Defaults to ``False``. diff --git a/slixmpp/xmlstream/handler/collector.py b/slixmpp/xmlstream/handler/collector.py index 8d012873..a5ee109c 100644 --- a/slixmpp/xmlstream/handler/collector.py +++ b/slixmpp/xmlstream/handler/collector.py @@ -4,11 +4,17 @@ # Part of Slixmpp: The Slick XMPP Library # :copyright: (c) 2012 Nathanael C. Fritz, Lance J.T. Stout # :license: MIT, see LICENSE for more details +from __future__ import annotations + import logging -from queue import Queue, Empty +from typing import List, Optional, TYPE_CHECKING +from slixmpp.xmlstream.stanzabase import StanzaBase from slixmpp.xmlstream.handler.base import BaseHandler +from slixmpp.xmlstream.matcher.base import MatcherBase +if TYPE_CHECKING: + from slixmpp.xmlstream.xmlstream import XMLStream log = logging.getLogger(__name__) @@ -27,35 +33,35 @@ class Collector(BaseHandler): :param stream: The :class:`~slixmpp.xmlstream.xmlstream.XMLStream` instance this handler should monitor. """ + _stanzas: List[StanzaBase] - def __init__(self, name, matcher, stream=None): + def __init__(self, name: str, matcher: MatcherBase, stream: Optional[XMLStream] = None): BaseHandler.__init__(self, name, matcher, stream=stream) - self._payload = Queue() + self._stanzas = [] - def prerun(self, payload): + def prerun(self, payload: StanzaBase) -> None: """Store the matched stanza when received during processing. :param payload: The matched - :class:`~slixmpp.xmlstream.stanzabase.ElementBase` object. + :class:`~slixmpp.xmlstream.stanzabase.StanzaBase` object. """ - self._payload.put(payload) + self._stanzas.append(payload) - def run(self, payload): + def run(self, payload: StanzaBase) -> None: """Do not process this handler during the main event loop.""" pass - def stop(self): + def stop(self) -> List[StanzaBase]: """ Stop collection of matching stanzas, and return the ones that have been stored so far. """ + stream_ref = self.stream + if stream_ref is None: + raise ValueError('stop() called without a stream!') + stream = stream_ref() + if stream is None: + raise ValueError('stop() called without a stream!') self._destroy = True - results = [] - try: - while True: - results.append(self._payload.get(False)) - except Empty: - pass - - self.stream().remove_handler(self.name) - return results + stream.remove_handler(self.name) + return self._stanzas diff --git a/slixmpp/xmlstream/handler/coroutine_callback.py b/slixmpp/xmlstream/handler/coroutine_callback.py index 6568ba9f..d41cd7ba 100644 --- a/slixmpp/xmlstream/handler/coroutine_callback.py +++ b/slixmpp/xmlstream/handler/coroutine_callback.py @@ -4,8 +4,19 @@ # Part of Slixmpp: The Slick XMPP Library # :copyright: (c) 2011 Nathanael C. Fritz # :license: MIT, see LICENSE for more details +from __future__ import annotations + +from asyncio import iscoroutinefunction, ensure_future +from typing import Optional, Callable, Awaitable, TYPE_CHECKING + +from slixmpp.xmlstream.stanzabase import StanzaBase from slixmpp.xmlstream.handler.base import BaseHandler -from slixmpp.xmlstream.asyncio import asyncio +from slixmpp.xmlstream.matcher.base import MatcherBase + +CoroutineFunction = Callable[[StanzaBase], Awaitable[None]] + +if TYPE_CHECKING: + from slixmpp.xmlstream.xmlstream import XMLStream class CoroutineCallback(BaseHandler): @@ -34,15 +45,20 @@ class CoroutineCallback(BaseHandler): instance this handler should monitor. """ - def __init__(self, name, matcher, pointer, once=False, - instream=False, stream=None): + _once: bool + _instream: bool + _pointer: CoroutineFunction + + def __init__(self, name: str, matcher: MatcherBase, + pointer: CoroutineFunction, once: bool = False, + instream: bool = False, stream: Optional[XMLStream] = None): BaseHandler.__init__(self, name, matcher, stream) - if not asyncio.iscoroutinefunction(pointer): + if not iscoroutinefunction(pointer): raise ValueError("Given function is not a coroutine") - async def pointer_wrapper(stanza, *args, **kwargs): + async def pointer_wrapper(stanza: StanzaBase) -> None: try: - await pointer(stanza, *args, **kwargs) + await pointer(stanza) except Exception as e: stanza.exception(e) @@ -50,29 +66,29 @@ class CoroutineCallback(BaseHandler): self._once = once self._instream = instream - def prerun(self, payload): + def prerun(self, payload: StanzaBase) -> None: """Execute the callback during stream processing, if the callback was created with ``instream=True``. :param payload: The matched - :class:`~slixmpp.xmlstream.stanzabase.ElementBase` object. + :class:`~slixmpp.xmlstream.stanzabase.StanzaBase` object. """ if self._once: self._destroy = True if self._instream: self.run(payload, True) - def run(self, payload, instream=False): + def run(self, payload: StanzaBase, instream: bool = False) -> None: """Execute the callback function with the matched stanza payload. :param payload: The matched - :class:`~slixmpp.xmlstream.stanzabase.ElementBase` object. + :class:`~slixmpp.xmlstream.stanzabase.StanzaBase` object. :param bool instream: Force the handler to execute during stream processing. This should only be used by :meth:`prerun()`. Defaults to ``False``. """ if not self._instream or instream: - asyncio.ensure_future(self._pointer(payload)) + ensure_future(self._pointer(payload)) if self._once: self._destroy = True del self._pointer diff --git a/slixmpp/xmlstream/handler/waiter.py b/slixmpp/xmlstream/handler/waiter.py index 758cd1f1..684bd32f 100644 --- a/slixmpp/xmlstream/handler/waiter.py +++ b/slixmpp/xmlstream/handler/waiter.py @@ -4,13 +4,19 @@ # Part of Slixmpp: The Slick XMPP Library # :copyright: (c) 2011 Nathanael C. Fritz # :license: MIT, see LICENSE for more details +from __future__ import annotations + import logging -import asyncio -from asyncio import Queue, wait_for, TimeoutError +from asyncio import Event, wait_for, TimeoutError +from typing import Optional, TYPE_CHECKING import slixmpp +from slixmpp.xmlstream.stanzabase import StanzaBase from slixmpp.xmlstream.handler.base import BaseHandler +from slixmpp.xmlstream.matcher.base import MatcherBase +if TYPE_CHECKING: + from slixmpp.xmlstream.xmlstream import XMLStream log = logging.getLogger(__name__) @@ -28,24 +34,27 @@ class Waiter(BaseHandler): :param stream: The :class:`~slixmpp.xmlstream.xmlstream.XMLStream` instance this handler should monitor. """ + _event: Event - def __init__(self, name, matcher, stream=None): + def __init__(self, name: str, matcher: MatcherBase, stream: Optional[XMLStream] = None): BaseHandler.__init__(self, name, matcher, stream=stream) - self._payload = Queue() + self._event = Event() - def prerun(self, payload): + def prerun(self, payload: StanzaBase) -> None: """Store the matched stanza when received during processing. :param payload: The matched - :class:`~slixmpp.xmlstream.stanzabase.ElementBase` object. + :class:`~slixmpp.xmlstream.stanzabase.StanzaBase` object. """ - self._payload.put_nowait(payload) + if not self._event.is_set(): + self._event.set() + self._payload = payload - def run(self, payload): + def run(self, payload: StanzaBase) -> None: """Do not process this handler during the main event loop.""" pass - async def wait(self, timeout=None): + async def wait(self, timeout: Optional[int] = None) -> Optional[StanzaBase]: """Block an event handler while waiting for a stanza to arrive. Be aware that this will impact performance if called from a @@ -59,17 +68,24 @@ class Waiter(BaseHandler): :class:`~slixmpp.xmlstream.xmlstream.XMLStream.response_timeout` value. """ + stream_ref = self.stream + if stream_ref is None: + raise ValueError('wait() called without a stream') + stream = stream_ref() + if stream is None: + raise ValueError('wait() called without a stream') if timeout is None: timeout = slixmpp.xmlstream.RESPONSE_TIMEOUT - stanza = None try: - stanza = await self._payload.get() + await wait_for( + self._event.wait(), timeout, loop=stream.loop + ) except TimeoutError: log.warning("Timed out waiting for %s", self.name) - self.stream().remove_handler(self.name) - return stanza + stream.remove_handler(self.name) + return self._payload - def check_delete(self): + def check_delete(self) -> bool: """Always remove waiters after use.""" return True diff --git a/slixmpp/xmlstream/handler/xmlcallback.py b/slixmpp/xmlstream/handler/xmlcallback.py index b44b2da1..31534033 100644 --- a/slixmpp/xmlstream/handler/xmlcallback.py +++ b/slixmpp/xmlstream/handler/xmlcallback.py @@ -4,6 +4,7 @@ # This file is part of Slixmpp. # See the file LICENSE for copying permission. from slixmpp.xmlstream.handler import Callback +from slixmpp.xmlstream.stanzabase import StanzaBase class XMLCallback(Callback): @@ -17,7 +18,7 @@ class XMLCallback(Callback): run -- Overrides Callback.run """ - def run(self, payload, instream=False): + def run(self, payload: StanzaBase, instream: bool = False) -> None: """ Execute the callback function with the matched stanza's XML contents, instead of the stanza itself. diff --git a/slixmpp/xmlstream/handler/xmlwaiter.py b/slixmpp/xmlstream/handler/xmlwaiter.py index 6eb6577e..65185a5f 100644 --- a/slixmpp/xmlstream/handler/xmlwaiter.py +++ b/slixmpp/xmlstream/handler/xmlwaiter.py @@ -3,6 +3,7 @@ # Copyright (C) 2010 Nathanael C. Fritz # This file is part of Slixmpp. # See the file LICENSE for copying permission. +from slixmpp.xmlstream.stanzabase import StanzaBase from slixmpp.xmlstream.handler import Waiter @@ -17,7 +18,7 @@ class XMLWaiter(Waiter): prerun -- Overrides Waiter.prerun """ - def prerun(self, payload): + def prerun(self, payload: StanzaBase) -> None: """ Store the XML contents of the stanza to return to the waiting event handler. diff --git a/slixmpp/xmlstream/matcher/base.py b/slixmpp/xmlstream/matcher/base.py index e2560d2b..552269c5 100644 --- a/slixmpp/xmlstream/matcher/base.py +++ b/slixmpp/xmlstream/matcher/base.py @@ -1,10 +1,13 @@ - # slixmpp.xmlstream.matcher.base # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ # Part of Slixmpp: The Slick XMPP Library # :copyright: (c) 2011 Nathanael C. Fritz # :license: MIT, see LICENSE for more details +from typing import Any +from slixmpp.xmlstream.stanzabase import StanzaBase + + class MatcherBase(object): """ @@ -15,10 +18,10 @@ class MatcherBase(object): :param criteria: Object to compare some aspect of a stanza against. """ - def __init__(self, criteria): + def __init__(self, criteria: Any): self._criteria = criteria - def match(self, xml): + def match(self, xml: StanzaBase) -> bool: """Check if a stanza matches the stored criteria. Meant to be overridden. diff --git a/slixmpp/xmlstream/matcher/id.py b/slixmpp/xmlstream/matcher/id.py index 44df2c18..e4e7ad4e 100644 --- a/slixmpp/xmlstream/matcher/id.py +++ b/slixmpp/xmlstream/matcher/id.py @@ -5,6 +5,7 @@ # :copyright: (c) 2011 Nathanael C. Fritz # :license: MIT, see LICENSE for more details from slixmpp.xmlstream.matcher.base import MatcherBase +from slixmpp.xmlstream.stanzabase import StanzaBase class MatcherId(MatcherBase): @@ -13,12 +14,13 @@ class MatcherId(MatcherBase): The ID matcher selects stanzas that have the same stanza 'id' interface value as the desired ID. """ + _criteria: str - def match(self, xml): + def match(self, xml: StanzaBase) -> bool: """Compare the given stanza's ``'id'`` attribute to the stored ``id`` value. - :param xml: The :class:`~slixmpp.xmlstream.stanzabase.ElementBase` + :param xml: The :class:`~slixmpp.xmlstream.stanzabase.StanzaBase` stanza to compare against. """ - return xml['id'] == self._criteria + return bool(xml['id'] == self._criteria) diff --git a/slixmpp/xmlstream/matcher/idsender.py b/slixmpp/xmlstream/matcher/idsender.py index 8f5d0303..9b12623d 100644 --- a/slixmpp/xmlstream/matcher/idsender.py +++ b/slixmpp/xmlstream/matcher/idsender.py @@ -4,7 +4,17 @@ # Part of Slixmpp: The Slick XMPP Library # :copyright: (c) 2011 Nathanael C. Fritz # :license: MIT, see LICENSE for more details + from slixmpp.xmlstream.matcher.base import MatcherBase +from slixmpp.xmlstream.stanzabase import StanzaBase +from slixmpp.jid import JID +from slixmpp.types import TypedDict + + +class CriteriaType(TypedDict): + self: JID + peer: JID + id: str class MatchIDSender(MatcherBase): @@ -14,12 +24,13 @@ class MatchIDSender(MatcherBase): interface value as the desired ID, and that the 'from' value is one of a set of approved entities that can respond to a request. """ + _criteria: CriteriaType - def match(self, xml): + def match(self, xml: StanzaBase) -> bool: """Compare the given stanza's ``'id'`` attribute to the stored ``id`` value, and verify the sender's JID. - :param xml: The :class:`~slixmpp.xmlstream.stanzabase.ElementBase` + :param xml: The :class:`~slixmpp.xmlstream.stanzabase.StanzaBase` stanza to compare against. """ diff --git a/slixmpp/xmlstream/matcher/many.py b/slixmpp/xmlstream/matcher/many.py index e8ad54e7..dae45463 100644 --- a/slixmpp/xmlstream/matcher/many.py +++ b/slixmpp/xmlstream/matcher/many.py @@ -3,7 +3,9 @@ # Copyright (C) 2010 Nathanael C. Fritz # This file is part of Slixmpp. # See the file LICENSE for copying permission. +from typing import Iterable from slixmpp.xmlstream.matcher.base import MatcherBase +from slixmpp.xmlstream.stanzabase import StanzaBase class MatchMany(MatcherBase): @@ -18,8 +20,9 @@ class MatchMany(MatcherBase): Methods: match -- Overrides MatcherBase.match. """ + _criteria: Iterable[MatcherBase] - def match(self, xml): + def match(self, xml: StanzaBase) -> bool: """ Match a stanza against multiple criteria. The match is successful if one of the criteria matches. diff --git a/slixmpp/xmlstream/matcher/stanzapath.py b/slixmpp/xmlstream/matcher/stanzapath.py index 1bf3fa8e..b7de3ee2 100644 --- a/slixmpp/xmlstream/matcher/stanzapath.py +++ b/slixmpp/xmlstream/matcher/stanzapath.py @@ -4,8 +4,9 @@ # Part of Slixmpp: The Slick XMPP Library # :copyright: (c) 2011 Nathanael C. Fritz # :license: MIT, see LICENSE for more details +from typing import cast, List from slixmpp.xmlstream.matcher.base import MatcherBase -from slixmpp.xmlstream.stanzabase import fix_ns +from slixmpp.xmlstream.stanzabase import fix_ns, StanzaBase class StanzaPath(MatcherBase): @@ -17,22 +18,28 @@ class StanzaPath(MatcherBase): :param criteria: Object to compare some aspect of a stanza against. """ - - def __init__(self, criteria): - self._criteria = fix_ns(criteria, split=True, - propagate_ns=False, - default_ns='jabber:client') + _criteria: List[str] + _raw_criteria: str + + def __init__(self, criteria: str): + self._criteria = cast( + List[str], + fix_ns( + criteria, split=True, propagate_ns=False, + default_ns='jabber:client' + ) + ) self._raw_criteria = criteria - def match(self, stanza): + def match(self, stanza: StanzaBase) -> bool: """ Compare a stanza against a "stanza path". A stanza path is similar to an XPath expression, but uses the stanza's interfaces and plugins instead of the underlying XML. See the documentation for the stanza - :meth:`~slixmpp.xmlstream.stanzabase.ElementBase.match()` method + :meth:`~slixmpp.xmlstream.stanzabase.StanzaBase.match()` method for more information. - :param stanza: The :class:`~slixmpp.xmlstream.stanzabase.ElementBase` + :param stanza: The :class:`~slixmpp.xmlstream.stanzabase.StanzaBase` stanza to compare against. """ return stanza.match(self._criteria) or stanza.match(self._raw_criteria) diff --git a/slixmpp/xmlstream/matcher/xmlmask.py b/slixmpp/xmlstream/matcher/xmlmask.py index d50b706e..b63e0f05 100644 --- a/slixmpp/xmlstream/matcher/xmlmask.py +++ b/slixmpp/xmlstream/matcher/xmlmask.py @@ -1,4 +1,3 @@ - # Slixmpp: The Slick XMPP Library # Copyright (C) 2010 Nathanael C. Fritz # This file is part of Slixmpp. @@ -6,8 +5,9 @@ import logging from xml.parsers.expat import ExpatError +from xml.etree.ElementTree import Element -from slixmpp.xmlstream.stanzabase import ET +from slixmpp.xmlstream.stanzabase import ET, StanzaBase from slixmpp.xmlstream.matcher.base import MatcherBase @@ -33,32 +33,33 @@ class MatchXMLMask(MatcherBase): :param criteria: Either an :class:`~xml.etree.ElementTree.Element` XML object or XML string to use as a mask. """ + _criteria: Element - def __init__(self, criteria, default_ns='jabber:client'): + def __init__(self, criteria: str, default_ns: str = 'jabber:client'): MatcherBase.__init__(self, criteria) if isinstance(criteria, str): - self._criteria = ET.fromstring(self._criteria) + self._criteria = ET.fromstring(criteria) self.default_ns = default_ns - def setDefaultNS(self, ns): + def setDefaultNS(self, ns: str) -> None: """Set the default namespace to use during comparisons. :param ns: The new namespace to use as the default. """ self.default_ns = ns - def match(self, xml): + def match(self, xml: StanzaBase) -> bool: """Compare a stanza object or XML object against the stored XML mask. Overrides MatcherBase.match. :param xml: The stanza object or XML object to compare against. """ - if hasattr(xml, 'xml'): - xml = xml.xml - return self._mask_cmp(xml, self._criteria, True) + real_xml = xml.xml + return self._mask_cmp(real_xml, self._criteria, True) - def _mask_cmp(self, source, mask, use_ns=False, default_ns='__no_ns__'): + def _mask_cmp(self, source: Element, mask: Element, use_ns: bool = False, + default_ns: str = '__no_ns__') -> bool: """Compare an XML object against an XML mask. :param source: The :class:`~xml.etree.ElementTree.Element` XML object @@ -75,13 +76,6 @@ class MatchXMLMask(MatcherBase): # If the element was not found. May happen during recursive calls. return False - # Convert the mask to an XML object if it is a string. - if not hasattr(mask, 'attrib'): - try: - mask = ET.fromstring(mask) - except ExpatError: - log.warning("Expat error: %s\nIn parsing: %s", '', mask) - mask_ns_tag = "{%s}%s" % (self.default_ns, mask.tag) if source.tag not in [mask.tag, mask_ns_tag]: return False diff --git a/slixmpp/xmlstream/matcher/xpath.py b/slixmpp/xmlstream/matcher/xpath.py index b7503b73..bd41b60a 100644 --- a/slixmpp/xmlstream/matcher/xpath.py +++ b/slixmpp/xmlstream/matcher/xpath.py @@ -4,7 +4,8 @@ # Part of Slixmpp: The Slick XMPP Library # :copyright: (c) 2011 Nathanael C. Fritz # :license: MIT, see LICENSE for more details -from slixmpp.xmlstream.stanzabase import ET, fix_ns +from typing import cast +from slixmpp.xmlstream.stanzabase import ET, fix_ns, StanzaBase from slixmpp.xmlstream.matcher.base import MatcherBase @@ -17,23 +18,23 @@ class MatchXPath(MatcherBase): If the value of :data:`IGNORE_NS` is set to ``True``, then XPath expressions will be matched without using namespaces. """ + _criteria: str - def __init__(self, criteria): - self._criteria = fix_ns(criteria) + def __init__(self, criteria: str): + self._criteria = cast(str, fix_ns(criteria)) - def match(self, xml): + def match(self, xml: StanzaBase) -> bool: """ Compare a stanza's XML contents to an XPath expression. If the value of :data:`IGNORE_NS` is set to ``True``, then XPath expressions will be matched without using namespaces. - :param xml: The :class:`~slixmpp.xmlstream.stanzabase.ElementBase` + :param xml: The :class:`~slixmpp.xmlstream.stanzabase.StanzaBase` stanza to compare against. """ - if hasattr(xml, 'xml'): - xml = xml.xml + real_xml = xml.xml x = ET.Element('x') - x.append(xml) + x.append(real_xml) return x.find(self._criteria) is not None -- cgit v1.2.3