GitLab

at 2021-11-03T00:44:47+01:00

Protocol: Add annotations

BaseModule: Add annotations

Blocking: Add annotations

 # 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)

 # 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

         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)

         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)

     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:

         jids.add(jid)

-

-    block, unblock = set(), set()

+    block: Set[JID] = set()

+    unblock: Set[JID] = set()

     if node.getName() == 'block':

         block = jids

     else:

     """

     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.

         if queryNS:

             self.setQueryNS(queryNS)

-    def getQuery(self):

+    def getQuery(self) -> Optional[Node]:

         """

         Return the IQ's child element if it exists, None otherwise.

         """

             return children[0]

         return None

-    def getQueryNS(self):

+    def getQueryNS(self) -> Optional[str]:

         """

         Return the namespace of the 'query' child element

         """

             return tag.getNamespace()

         return None

-    def getQuerynode(self):

+    def getQuerynode(self) -> Optional[str]:

         """

         Return the 'node' attribute value of the 'query' child element

         """

             return tag.getAttr('node')

         return None

-    def getQueryPayload(self):

+    def getQueryPayload(self) -> Optional[list[Union[Node, str]]]:

         """

         Return the 'query' child element payload

         """

             return tag.getPayload()

         return None

-    def getQueryChildren(self):

+    def getQueryChildren(self) -> Optional[list[Node]]:

         """

         Return the 'query' child element child nodes

         """

             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

                     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).

             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.

         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()})

 from __future__ import annotations

+import typing

 from typing import Any

 from typing import Optional

 from typing import Set

 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

         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

+

+from typing import Union

+from typing import Literal

+

+BlockingReportValues = Union[Literal['spam'], Literal['abuse']]

Philipp Hörist pushed to branch master at gajim / python-nbxmpp

Commits:

5 changed files:

Changes: