summaryrefslogtreecommitdiff
path: root/slixmpp/stanza
diff options
context:
space:
mode:
Diffstat (limited to 'slixmpp/stanza')
-rw-r--r--slixmpp/stanza/error.py27
-rw-r--r--slixmpp/stanza/handshake.py3
-rw-r--r--slixmpp/stanza/iq.py4
-rw-r--r--slixmpp/stanza/message.py12
-rw-r--r--slixmpp/stanza/presence.py17
-rw-r--r--slixmpp/stanza/stream_error.py12
-rw-r--r--slixmpp/stanza/stream_features.py7
7 files changed, 47 insertions, 35 deletions
diff --git a/slixmpp/stanza/error.py b/slixmpp/stanza/error.py
index 54ace5a8..76c6b9cc 100644
--- a/slixmpp/stanza/error.py
+++ b/slixmpp/stanza/error.py
@@ -1,8 +1,9 @@
-
# Slixmpp: The Slick XMPP Library
# Copyright (C) 2010 Nathanael C. Fritz
# This file is part of Slixmpp.
# See the file LICENSE for copying permission.
+from __future__ import annotations
+from typing import Optional, Dict, Type, ClassVar
from slixmpp.xmlstream import ElementBase, ET
@@ -49,10 +50,10 @@ class Error(ElementBase):
name = 'error'
plugin_attrib = 'error'
interfaces = {'code', 'condition', 'text', 'type',
- 'gone', 'redirect', 'by'}
+ 'gone', 'redirect', 'by'}
sub_interfaces = {'text'}
- plugin_attrib_map = {}
- plugin_tag_map = {}
+ plugin_attrib_map: ClassVar[Dict[str, Type[ElementBase]]] = {}
+ plugin_tag_map: ClassVar[Dict[str, Type[ElementBase]]] = {}
conditions = {'bad-request', 'conflict', 'feature-not-implemented',
'forbidden', 'gone', 'internal-server-error',
'item-not-found', 'jid-malformed', 'not-acceptable',
@@ -62,10 +63,10 @@ class Error(ElementBase):
'remote-server-timeout', 'resource-constraint',
'service-unavailable', 'subscription-required',
'undefined-condition', 'unexpected-request'}
- condition_ns = 'urn:ietf:params:xml:ns:xmpp-stanzas'
+ condition_ns: str = 'urn:ietf:params:xml:ns:xmpp-stanzas'
types = {'cancel', 'continue', 'modify', 'auth', 'wait'}
- def setup(self, xml=None):
+ def setup(self, xml: Optional[ET.Element] = None):
"""
Populate the stanza object using an optional XML object.
@@ -82,9 +83,11 @@ class Error(ElementBase):
self['type'] = 'cancel'
self['condition'] = 'feature-not-implemented'
if self.parent is not None:
- self.parent()['type'] = 'error'
+ parent = self.parent()
+ if parent:
+ parent['type'] = 'error'
- def get_condition(self):
+ def get_condition(self) -> str:
"""Return the condition element's name."""
for child in self.xml:
if "{%s}" % self.condition_ns in child.tag:
@@ -93,7 +96,7 @@ class Error(ElementBase):
return cond
return ''
- def set_condition(self, value):
+ def set_condition(self, value: str) -> Error:
"""
Set the tag name of the condition element.
@@ -105,7 +108,7 @@ class Error(ElementBase):
self.xml.append(ET.Element("{%s}%s" % (self.condition_ns, value)))
return self
- def del_condition(self):
+ def del_condition(self) -> Error:
"""Remove the condition element."""
for child in self.xml:
if "{%s}" % self.condition_ns in child.tag:
@@ -139,14 +142,14 @@ class Error(ElementBase):
def get_redirect(self):
return self._get_sub_text('{%s}redirect' % self.condition_ns, '')
- def set_gone(self, value):
+ def set_gone(self, value: str):
if value:
del self['condition']
return self._set_sub_text('{%s}gone' % self.condition_ns, value)
elif self['condition'] == 'gone':
del self['condition']
- def set_redirect(self, value):
+ def set_redirect(self, value: str):
if value:
del self['condition']
ns = self.condition_ns
diff --git a/slixmpp/stanza/handshake.py b/slixmpp/stanza/handshake.py
index c58f69aa..70f890be 100644
--- a/slixmpp/stanza/handshake.py
+++ b/slixmpp/stanza/handshake.py
@@ -4,6 +4,7 @@
# See the file LICENSE for copying permission.
from slixmpp.xmlstream import StanzaBase
+from typing import Optional
class Handshake(StanzaBase):
@@ -18,7 +19,7 @@ class Handshake(StanzaBase):
def set_value(self, value: str):
self.xml.text = value
- def get_value(self) -> str:
+ def get_value(self) -> Optional[str]:
return self.xml.text
def del_value(self):
diff --git a/slixmpp/stanza/iq.py b/slixmpp/stanza/iq.py
index 044c9df8..34e56f60 100644
--- a/slixmpp/stanza/iq.py
+++ b/slixmpp/stanza/iq.py
@@ -3,10 +3,10 @@
# Copyright (C) 2010 Nathanael C. Fritz
# This file is part of Slixmpp.
# See the file LICENSE for copying permission.
+import asyncio
from slixmpp.stanza.rootstanza import RootStanza
from slixmpp.xmlstream import StanzaBase, ET
-from slixmpp.xmlstream.handler import Waiter, Callback, CoroutineCallback
-from slixmpp.xmlstream.asyncio import asyncio
+from slixmpp.xmlstream.handler import Callback, CoroutineCallback
from slixmpp.xmlstream.matcher import MatchIDSender, MatcherId
from slixmpp.exceptions import IqTimeout, IqError
diff --git a/slixmpp/stanza/message.py b/slixmpp/stanza/message.py
index debfb380..50d32ff0 100644
--- a/slixmpp/stanza/message.py
+++ b/slixmpp/stanza/message.py
@@ -61,8 +61,10 @@ class Message(RootStanza):
"""
StanzaBase.__init__(self, *args, **kwargs)
if not recv and self['id'] == '':
- if self.stream is not None and self.stream.use_message_ids:
- self['id'] = self.stream.new_id()
+ if self.stream:
+ use_ids = getattr(self.stream, 'use_message_ids', None)
+ if use_ids:
+ self['id'] = self.stream.new_id()
else:
del self['origin_id']
@@ -93,8 +95,10 @@ class Message(RootStanza):
self.xml.attrib['id'] = value
- if self.stream and not self.stream.use_origin_id:
- return None
+ if self.stream:
+ use_orig_ids = getattr(self.stream, 'use_origin_id', None)
+ if not use_orig_ids:
+ return None
sub = self.xml.find(ORIGIN_NAME)
if sub is not None:
diff --git a/slixmpp/stanza/presence.py b/slixmpp/stanza/presence.py
index 022e7133..d77ce1e4 100644
--- a/slixmpp/stanza/presence.py
+++ b/slixmpp/stanza/presence.py
@@ -1,4 +1,3 @@
-
# Slixmpp: The Slick XMPP Library
# Copyright (C) 2010 Nathanael C. Fritz
# This file is part of Slixmpp.
@@ -61,7 +60,7 @@ class Presence(RootStanza):
'subscribed', 'unsubscribe', 'unsubscribed'}
showtypes = {'dnd', 'chat', 'xa', 'away'}
- def __init__(self, *args, recv=False, **kwargs):
+ def __init__(self, *args, recv: bool = False, **kwargs):
"""
Initialize a new <presence /> stanza with an optional 'id' value.
@@ -69,10 +68,12 @@ class Presence(RootStanza):
"""
StanzaBase.__init__(self, *args, **kwargs)
if not recv and self['id'] == '':
- if self.stream is not None and self.stream.use_presence_ids:
- self['id'] = self.stream.new_id()
+ if self.stream:
+ use_ids = getattr(self.stream, 'use_presence_ids', None)
+ if use_ids:
+ self['id'] = self.stream.new_id()
- def set_show(self, show):
+ def set_show(self, show: str):
"""
Set the value of the <show> element.
@@ -84,7 +85,7 @@ class Presence(RootStanza):
self._set_sub_text('show', text=show)
return self
- def get_type(self):
+ def get_type(self) -> str:
"""
Return the value of the <presence> stanza's type attribute, or
the value of the <show> element if valid.
@@ -96,7 +97,7 @@ class Presence(RootStanza):
out = 'available'
return out
- def set_type(self, value):
+ def set_type(self, value: str):
"""
Set the type attribute's value, and the <show> element
if applicable.
@@ -119,7 +120,7 @@ class Presence(RootStanza):
self._del_attr('type')
self._del_sub('show')
- def set_priority(self, value):
+ def set_priority(self, value: int):
"""
Set the entity's priority value. Some server use priority to
determine message routing behavior.
diff --git a/slixmpp/stanza/stream_error.py b/slixmpp/stanza/stream_error.py
index 0e728c8e..d0eadd5b 100644
--- a/slixmpp/stanza/stream_error.py
+++ b/slixmpp/stanza/stream_error.py
@@ -4,7 +4,8 @@
# This file is part of Slixmpp.
# See the file LICENSE for copying permission.
from slixmpp.stanza.error import Error
-from slixmpp.xmlstream import StanzaBase
+from slixmpp.xmlstream import StanzaBase, ET
+from typing import Optional, Dict, Union
class StreamError(Error, StanzaBase):
@@ -62,19 +63,20 @@ class StreamError(Error, StanzaBase):
'system-shutdown', 'undefined-condition', 'unsupported-encoding',
'unsupported-feature', 'unsupported-stanza-type',
'unsupported-version'}
- condition_ns = 'urn:ietf:params:xml:ns:xmpp-streams'
+ condition_ns: str = 'urn:ietf:params:xml:ns:xmpp-streams'
- def get_see_other_host(self):
+ def get_see_other_host(self) -> Union[str, Dict[str, str]]:
ns = self.condition_ns
return self._get_sub_text('{%s}see-other-host' % ns, '')
- def set_see_other_host(self, value):
+ def set_see_other_host(self, value: str) -> Optional[ET.Element]:
if value:
del self['condition']
ns = self.condition_ns
return self._set_sub_text('{%s}see-other-host' % ns, value)
elif self['condition'] == 'see-other-host':
del self['condition']
+ return None
- def del_see_other_host(self):
+ def del_see_other_host(self) -> None:
self._del_sub('{%s}see-other-host' % self.condition_ns)
diff --git a/slixmpp/stanza/stream_features.py b/slixmpp/stanza/stream_features.py
index 7362f17b..a3e05a1a 100644
--- a/slixmpp/stanza/stream_features.py
+++ b/slixmpp/stanza/stream_features.py
@@ -3,7 +3,8 @@
# Copyright (C) 2010 Nathanael C. Fritz
# This file is part of Slixmpp.
# See the file LICENSE for copying permission.
-from slixmpp.xmlstream import StanzaBase
+from slixmpp.xmlstream import StanzaBase, ElementBase
+from typing import ClassVar, Dict, Type
class StreamFeatures(StanzaBase):
@@ -15,8 +16,8 @@ class StreamFeatures(StanzaBase):
namespace = 'http://etherx.jabber.org/streams'
interfaces = {'features', 'required', 'optional'}
sub_interfaces = interfaces
- plugin_tag_map = {}
- plugin_attrib_map = {}
+ plugin_attrib_map: ClassVar[Dict[str, Type[ElementBase]]] = {}
+ plugin_tag_map: ClassVar[Dict[str, Type[ElementBase]]] = {}
def setup(self, xml):
StanzaBase.setup(self, xml)
generated by cgit v1.2.3 (git 2.25.1) at 2024-05-03 07:37:57 +0000