Philipp Hörist pushed to branch master at gajim / python-nbxmpp
Commits: 0c23dfbc by lovetox at 2021-11-03T00:44:47+01:00 Protocol: Add annotations
- - - - - c3eb064b by lovetox at 2021-11-03T00:45:16+01:00 BaseModule: Add annotations
- - - - - 1550765b by lovetox at 2021-11-03T00:46:22+01:00 Blocking: Add annotations
- - - - -
5 changed files:
- nbxmpp/modules/base.py - nbxmpp/modules/blocking.py - nbxmpp/protocol.py - nbxmpp/structs.py - + nbxmpp/types.py
Changes:
===================================== nbxmpp/modules/base.py ===================================== @@ -15,21 +15,29 @@ # You should have received a copy of the GNU General Public License # along with this program; If not, see http://www.gnu.org/licenses/.
+from __future__ import annotations + +from typing import TYPE_CHECKING +from typing import Any + import logging
from nbxmpp.util import LogAdapter
+if TYPE_CHECKING: + from nbxmpp.client import Client +
class BaseModule:
- _depends = {} + _depends: dict[str, str] = {}
- def __init__(self, client): + def __init__(self, client: Client): logger_name = 'nbxmpp.m.%s' % self.__class__.__name__.lower() self._log = LogAdapter(logging.getLogger(logger_name), {'context': client.log_context})
- def __getattr__(self, name): + def __getattr__(self, name: str) -> Any: if name not in self._depends: raise AttributeError('Unknown method: %s' % name)
===================================== nbxmpp/modules/blocking.py ===================================== @@ -15,20 +15,35 @@ # You should have received a copy of the GNU General Public License # along with this program; If not, see http://www.gnu.org/licenses/.
+from __future__ import annotations + +from typing import Literal, Optional +from typing import Union +from typing import Set +from typing import TYPE_CHECKING + from nbxmpp.namespaces import Namespace +from nbxmpp.simplexml import Node from nbxmpp.protocol import Iq from nbxmpp.protocol import JID -from nbxmpp.modules.base import BaseModule from nbxmpp.errors import StanzaError from nbxmpp.errors import MalformedStanzaError from nbxmpp.task import iq_request_task -from nbxmpp.structs import BlockingPush from nbxmpp.structs import StanzaHandler +from nbxmpp.structs import BlockingProperties +from nbxmpp.structs import BlockingPush +from nbxmpp.types import BlockingReportValues +from nbxmpp.modules.base import BaseModule from nbxmpp.modules.util import process_response
+if TYPE_CHECKING: + from nbxmpp.client import Client + + + class Blocking(BaseModule): - def __init__(self, client): + def __init__(self, client: Client): BaseModule.__init__(self, client)
self._client = client @@ -66,21 +81,27 @@ class Blocking(BaseModule): yield blocked
@iq_request_task - def block(self, jids, report=None): + def block(self, + jids: list[JID], + report: Optional[BlockingReportValues] = None): + _task = yield
response = yield _make_block_request(jids, report) yield process_response(response)
@iq_request_task - def unblock(self, jids): + def unblock(self, jids: list[JID]): _task = yield
response = yield _make_unblock_request(jids) yield process_response(response)
@staticmethod - def _process_blocking_push(client, stanza, properties): + def _process_blocking_push(client: Client, + stanza: Iq, + properties: BlockingProperties): + unblock = stanza.getTag('unblock', namespace=Namespace.BLOCKING) if unblock is not None: properties.blocking = _parse_push(unblock) @@ -93,17 +114,19 @@ class Blocking(BaseModule): client.send_stanza(reply)
-def _make_blocking_list_request(): +def _make_blocking_list_request() -> Iq: iq = Iq('get', Namespace.BLOCKING) iq.setQuery('blocklist') return iq
-def _make_block_request(jids, report): +def _make_block_request(jids: list[JID], + report: Optional[BlockingReportValues]) -> Iq: + iq = Iq('set', Namespace.BLOCKING) query = iq.setQuery(name='block') for jid in jids: - item = query.addChild(name='item', attrs={'jid': jid}) + item = query.addChild(name='item', attrs={'jid': str(jid)}) if report in ('spam', 'abuse'): action = item.addChild(name='report', namespace=Namespace.REPORTING) @@ -111,20 +134,20 @@ def _make_block_request(jids, report): return iq
-def _make_unblock_request(jids): +def _make_unblock_request(jids: list[JID]) -> Iq: iq = Iq('set', Namespace.BLOCKING) query = iq.setQuery(name='unblock') for jid in jids: - query.addChild(name='item', attrs={'jid': jid}) + query.addChild(name='item', attrs={'jid': str(jid)}) return iq
-def _parse_push(node): +def _parse_push(node: Node) -> BlockingPush: items = node.getTags('item') if not items: return BlockingPush(block=set(), unblock=set(), unblock_all=True)
- jids = set() + jids: Set[JID] = set() for item in items: jid = item.getAttr('jid') if not jid: @@ -137,8 +160,8 @@ def _parse_push(node):
jids.add(jid)
- - block, unblock = set(), set() + block: Set[JID] = set() + unblock: Set[JID] = set() if node.getName() == 'block': block = jids else:
===================================== nbxmpp/protocol.py ===================================== @@ -1387,14 +1387,14 @@ class Iq(Protocol): """
def __init__(self, - typ=None, - queryNS=None, - attrs=None, - to=None, - frm=None, - payload=None, - xmlns=Namespace.CLIENT, - node=None): + typ: Optional[str] = None, + queryNS: Optional[str] = None, + attrs: Optional[dict[str, str]] = None, + to: Optional[Union[JID, str]] = None, + frm: Optional[Union[JID, str]] = None, + payload: Optional[list[Union[Node, str]]] = None, + xmlns: str = Namespace.CLIENT, + node: Optional[Node] = None): """ You can specify type, query namespace any additional attributes, recipient of the iq, sender of the iq, any additional payload (f.e. @@ -1416,7 +1416,7 @@ class Iq(Protocol): if queryNS: self.setQueryNS(queryNS)
- def getQuery(self): + def getQuery(self) -> Optional[Node]: """ Return the IQ's child element if it exists, None otherwise. """ @@ -1426,7 +1426,7 @@ class Iq(Protocol): return children[0] return None
- def getQueryNS(self): + def getQueryNS(self) -> Optional[str]: """ Return the namespace of the 'query' child element """ @@ -1435,7 +1435,7 @@ class Iq(Protocol): return tag.getNamespace() return None
- def getQuerynode(self): + def getQuerynode(self) -> Optional[str]: """ Return the 'node' attribute value of the 'query' child element """ @@ -1444,7 +1444,7 @@ class Iq(Protocol): return tag.getAttr('node') return None
- def getQueryPayload(self): + def getQueryPayload(self) -> Optional[list[Union[Node, str]]]: """ Return the 'query' child element payload """ @@ -1453,7 +1453,7 @@ class Iq(Protocol): return tag.getPayload() return None
- def getQueryChildren(self): + def getQueryChildren(self) -> Optional[list[Node]]: """ Return the 'query' child element child nodes """ @@ -1462,7 +1462,7 @@ class Iq(Protocol): return tag.getChildren() return None
- def getQueryChild(self, name=None): + def getQueryChild(self, name: Optional[str] = None) -> Optional[Node]: """ Return the 'query' child element with name, or the first element which is not an error element @@ -1479,7 +1479,7 @@ class Iq(Protocol): return node return None
- def setQuery(self, name=None): + def setQuery(self, name: Optional[str] = None) -> Node: """ Change the name of the query node, creating it if needed. Keep the existing name if none is given (use 'query' if it's a creation). @@ -1492,25 +1492,25 @@ class Iq(Protocol): query.setName(name) return query
- def setQueryNS(self, namespace): + def setQueryNS(self, namespace: str) -> None: """ Set the namespace of the 'query' child element """ self.setQuery().setNamespace(namespace)
- def setQueryPayload(self, payload): + def setQueryPayload(self, payload: list[Union[Node, str]]): """ Set the 'query' child element payload """ self.setQuery().setPayload(payload)
- def setQuerynode(self, node): + def setQuerynode(self, node: str) -> None: """ Set the 'node' attribute value of the 'query' child element """ self.setQuery().setAttr('node', node)
- def buildReply(self, typ): + def buildReply(self, typ: str) -> Iq: """ Build and return another Iq object of specified type. The to, from and query child node of new Iq are pre-set as reply to this Iq. @@ -1522,7 +1522,7 @@ class Iq(Protocol): iq.setQuery(self.getQuery().getName()).setNamespace(self.getQueryNS()) return iq
- def buildSimpleReply(self, typ): + def buildSimpleReply(self, typ: str) -> Iq: return Iq(typ, to=self.getFrom(), attrs={'id': self.getID()})
===================================== nbxmpp/structs.py ===================================== @@ -17,6 +17,7 @@
from __future__ import annotations
+import typing from typing import Any from typing import Optional from typing import Set @@ -36,7 +37,8 @@ from nbxmpp.simplexml import Node from nbxmpp.namespaces import Namespace from nbxmpp.protocol import Protocol from nbxmpp.protocol import JID -from nbxmpp.const import AdHocAction, IqType +from nbxmpp.const import AdHocAction +from nbxmpp.const import IqType from nbxmpp.const import AdHocNoteType from nbxmpp.const import MessageType from nbxmpp.const import AvatarState @@ -1119,6 +1121,24 @@ class IqProperties: return self.roster is not None
+class IqPropertiesBase(typing.Protocol): + own_jid: JID + type: IqType + jid: JID + id: str + error: Optional[Any] + query: Optional[Node] + payload: Optional[Node] + + +class BlockingProperties(IqPropertiesBase): + + blocking: BlockingPush + + @property + def is_blocking(self) -> bool: ... + + @dataclass class PresenceProperties: own_jid: JID
===================================== nbxmpp/types.py ===================================== @@ -0,0 +1,5 @@ + +from typing import Union +from typing import Literal + +BlockingReportValues = Union[Literal['spam'], Literal['abuse']]
View it on GitLab: https://dev.gajim.org/gajim/python-nbxmpp/-/compare/07d5066463e8cdbcfe15636f...