Autobahn WebSocket Protocol - Improve typing

justfile

@@ -702,17 +702,15 @@ check-typing venv="": (install venv)
         --ignore unresolved-attribute \
         --ignore unresolved-reference \
         --ignore possibly-missing-attribute \
-        --ignore possibly-missing-import \
         --ignore call-non-callable \
         --ignore invalid-assignment \
         --ignore invalid-argument-type \
-        --ignore invalid-return-type \
         --ignore invalid-method-override \
         --ignore invalid-type-form \
         --ignore unsupported-operator \
         --ignore too-many-positional-arguments \
         --ignore unknown-argument \
-        --ignore non-subscriptable \
+        --ignore not-subscriptable \
         --ignore not-iterable \
         --ignore no-matching-overload \
         --ignore conflicting-declarations \

src/autobahn/twisted/websocket.py

@@ -25,7 +25,7 @@
 ###############################################################################
 
 from base64 import b64decode, b64encode
-from typing import Optional
+from typing import Any, Optional
 
 import txaio
 from zope.interface import implementer
@@ -79,14 +79,14 @@
 )
 
 
-def create_client_agent(reactor):
+def create_client_agent(reactor) -> "_TwistedWebSocketClientAgent":
     """
     :returns: an instance implementing IWebSocketClientAgent
     """
     return _TwistedWebSocketClientAgent(reactor)
 
 
-def check_transport_config(transport_config):
+def check_transport_config(transport_config: str) -> None:
     """
     raises a ValueError if `transport_config` is invalid
     """
@@ -107,7 +107,7 @@ def check_transport_config(transport_config):
     return None
 
 
-def check_client_options(options):
+def check_client_options(options: dict[str, Any]) -> None:
     """
     raises a ValueError if `options` is invalid
     """
@@ -264,7 +264,7 @@ class WebSocketAdapterProtocol(twisted.internet.protocol.Protocol):
     peer: Optional[str] = None
     is_server: Optional[bool] = None
 
-    def connectionMade(self):
+    def connectionMade(self) -> None:
         # Twisted networking framework entry point, called by Twisted
         # when the connection is established (either a client or a server)
 
@@ -296,7 +296,7 @@ def connectionMade(self):
             peer=hlval(self.peer),
         )
 
-    def connectionLost(self, reason: Failure = connectionDone):
+    def connectionLost(self, reason: Failure = connectionDone) -> None:
         # Twisted networking framework entry point, called by Twisted
         # when the connection is lost (either a client or a server)
 
@@ -352,7 +352,7 @@ def connectionLost(self, reason: Failure = connectionDone):
                 reason=reason,
             )
 
-    def dataReceived(self, data: bytes):
+    def dataReceived(self, data: bytes) -> None:
         self.log.debug(
             '{func} received {data_len} bytes for peer="{peer}"',
             func=hltype(self.dataReceived),
@@ -363,14 +363,14 @@ def dataReceived(self, data: bytes):
         # bytes received from Twisted, forward to the networking framework independent code for websocket
         self._dataReceived(data)
 
-    def _closeConnection(self, abort=False):
+    def _closeConnection(self, abort: bool=False) -> None:
         if abort and hasattr(self.transport, "abortConnection"):
             self.transport.abortConnection()
         else:
             # e.g. ProcessProtocol lacks abortConnection()
             self.transport.loseConnection()
 
-    def _onOpen(self):
+    def _onOpen(self) -> None:
         if self._transport_details.is_secure:
             # now that the TLS opening handshake is complete, the actual TLS channel ID
             # will be available. make sure to set it!
@@ -383,37 +383,37 @@ def _onOpen(self):
 
         self.onOpen()
 
-    def _onMessageBegin(self, isBinary):
+    def _onMessageBegin(self, isBinary: bool) -> None:
         self.onMessageBegin(isBinary)
 
-    def _onMessageFrameBegin(self, length):
+    def _onMessageFrameBegin(self, length: int) -> None:
         self.onMessageFrameBegin(length)
 
-    def _onMessageFrameData(self, payload):
+    def _onMessageFrameData(self, payload) -> None:
         self.onMessageFrameData(payload)
 
-    def _onMessageFrameEnd(self):
+    def _onMessageFrameEnd(self) -> None:
         self.onMessageFrameEnd()
 
-    def _onMessageFrame(self, payload):
+    def _onMessageFrame(self, payload) -> None:
         self.onMessageFrame(payload)
 
-    def _onMessageEnd(self):
+    def _onMessageEnd(self) -> None:
         self.onMessageEnd()
 
-    def _onMessage(self, payload, isBinary):
+    def _onMessage(self, payload, isBinary: bool) -> None:
         self.onMessage(payload, isBinary)
 
-    def _onPing(self, payload):
+    def _onPing(self, payload) -> None:
         self.onPing(payload)
 
-    def _onPong(self, payload):
+    def _onPong(self, payload) -> None:
         self.onPong(payload)
 
-    def _onClose(self, wasClean, code, reason):
+    def _onClose(self, wasClean: bool, code, reason) -> None:
         self.onClose(wasClean, code, reason)
 
-    def registerProducer(self, producer, streaming):
+    def registerProducer(self, producer, streaming) -> None:
         """
         Register a Twisted producer with this protocol.
 
@@ -424,7 +424,7 @@ def registerProducer(self, producer, streaming):
         """
         self.transport.registerProducer(producer, streaming)
 
-    def unregisterProducer(self):
+    def unregisterProducer(self) -> None:
         """
         Unregister Twisted producer with this protocol.
         """

src/autobahn/wamp/message.py

@@ -28,7 +28,7 @@
 import re
 import textwrap
 from pprint import pformat
-from typing import Any, Dict, Optional
+from typing import Any, Dict, Literal, Optional, overload
 
 import autobahn
 from autobahn.util import hlval
@@ -272,14 +272,38 @@ def identify_realm_name_category(value: Any) -> Optional[str]:
         return None
 
 
+@overload
+def check_or_raise_uri(
+    value: Any,
+    message: str,
+    strict: bool,
+    allow_empty_components: bool,
+    allow_last_empty: bool,
+    allow_none: Literal[True],
+) -> str | None:
+    pass
+
+
+@overload
 def check_or_raise_uri(
     value: Any,
     message: str = "WAMP message invalid",
     strict: bool = False,
     allow_empty_components: bool = False,
     allow_last_empty: bool = False,
-    allow_none: bool = False,
+    allow_none: Literal[False] = False,
 ) -> str:
+    pass
+
+
+def check_or_raise_uri(
+    value: Any,
+    message: str = "WAMP message invalid",
+    strict: bool = False,
+    allow_empty_components: bool = False,
+    allow_last_empty: bool = False,
+    allow_none: bool = False,
+) -> str | None:
     """
     Check a value for being a valid WAMP URI.
 

src/autobahn/wamp/types.py

@@ -2113,7 +2113,7 @@ def is_secure(self, value: Optional[bool]):
         self._is_secure = value
 
     @property
-    def channel_id(self) -> Dict[str, bytes]:
+    def channel_id(self) -> Dict[str, bytes] | None:
         """
         If this transport runs over a secure underlying connection, e.g. TLS,
         return a map of channel binding by binding type.
@@ -2161,7 +2161,7 @@ def channel_id(self, value: Dict[str, bytes]):
         self._channel_id = value
 
     @property
-    def peer_cert(self) -> Dict[str, Any]:
+    def peer_cert(self) -> Dict[str, Any] | None:
         """
         If this transport is using TLS and the TLS peer has provided a valid certificate,
         this attribute returns the peer certificate.
@@ -2200,7 +2200,7 @@ def websocket_extensions_in_use(self, value: Optional[List[str]]):
         self._websocket_extensions_in_use = value
 
     @property
-    def http_headers_received(self) -> Dict[str, Any]:
+    def http_headers_received(self) -> Dict[str, Any] | None:
         """
         If the underlying connection uses a regular HTTP based WebSocket opening handshake,
         the HTTP request headers as received from the client on this connection.
@@ -2212,7 +2212,7 @@ def http_headers_received(self, value: Dict[str, Any]):
         self._http_headers_received = value
 
     @property
-    def http_headers_sent(self) -> Dict[str, Any]:
+    def http_headers_sent(self) -> Dict[str, Any] | None:
         """
         If the underlying connection uses a regular HTTP based WebSocket opening handshake,
         the HTTP response headers as sent from the server on this connection.