Added explicit typing for socketpool and socket with shared interface

Author: michalpokusa

adafruit_httpserver/interfaces.py

@@ -8,11 +8,70 @@
 """
 
 try:
-    from typing import List, Dict, Union, Any
+    from typing import List, Tuple, Dict, Union, Any
 except ImportError:
     pass
 
 
+class _ISocket:  # pylint: disable=missing-function-docstring,no-self-use,unused-argument
+    """A class for typing necessary methods for a socket object."""
+
+    def accept(self) -> Tuple["_ISocket", Tuple[str, int]]:
+        ...
+
+    def bind(self, address: Tuple[str, int]) -> None:
+        ...
+
+    def setblocking(self, flag: bool) -> None:
+        ...
+
+    def settimeout(self, value: "Union[float, None]") -> None:
+        ...
+
+    def setsockopt(self, level: int, optname: int, value: int) -> None:
+        ...
+
+    def listen(self, backlog: int) -> None:
+        ...
+
+    def send(self, data: bytes) -> int:
+        ...
+
+    def recv_into(self, buffer: memoryview, nbytes: int) -> int:
+        ...
+
+    def close(self) -> None:
+        ...
+
+
+class _ISocketPool:  # pylint: disable=missing-function-docstring,no-self-use,unused-argument
+    """A class to typing necessary methods and properties for a socket pool object."""
+
+    AF_INET: int
+    SO_REUSEADDR: int
+    SOCK_STREAM: int
+    SOL_SOCKET: int
+
+    def socket(  # pylint: disable=redefined-builtin
+        self,
+        family: int = ...,
+        type: int = ...,
+        proto: int = ...,
+    ) -> _ISocket:
+        ...
+
+    def getaddrinfo(  # pylint: disable=redefined-builtin,too-many-arguments
+        self,
+        host: str,
+        port: int,
+        family: int = ...,
+        type: int = ...,
+        proto: int = ...,
+        flags: int = ...,
+    ) -> Tuple[int, int, int, str, Tuple[str, int]]:
+        ...
+
+
 class _IFieldStorage:
     """Interface with shared methods for QueryParams, FormData and Headers."""
 

adafruit_httpserver/request.py

@@ -9,8 +9,6 @@
 
 try:
     from typing import List, Dict, Tuple, Union, Any, TYPE_CHECKING
-    from socket import socket
-    from socketpool import SocketPool
 
     if TYPE_CHECKING:
         from .server import Server
@@ -20,7 +18,7 @@
 import json
 
 from .headers import Headers
-from .interfaces import _IFieldStorage, _IXSSSafeFieldStorage
+from .interfaces import _ISocket, _IFieldStorage, _IXSSSafeFieldStorage
 from .methods import POST, PUT, PATCH, DELETE
 
 
@@ -274,7 +272,7 @@ class Request:  # pylint: disable=too-many-instance-attributes
     Server object that received the request.
     """
 
-    connection: Union["SocketPool.Socket", "socket.socket"]
+    connection: _ISocket
     """
     Socket object used to send and receive data on the connection.
     """
@@ -325,7 +323,7 @@ class Request:  # pylint: disable=too-many-instance-attributes
     def __init__(
         self,
         server: "Server",
-        connection: Union["SocketPool.Socket", "socket.socket"],
+        connection: _ISocket,
         client_address: Tuple[str, int],
         raw_request: bytes = None,
     ) -> None:

adafruit_httpserver/response.py

@@ -9,8 +9,6 @@
 
 try:
     from typing import Optional, Dict, Union, Tuple, Generator, Any
-    from socket import socket
-    from socketpool import SocketPool
 except ImportError:
     pass
 
@@ -47,6 +45,7 @@
     PERMANENT_REDIRECT_308,
 )
 from .headers import Headers
+from .interfaces import _ISocket
 
 
 class Response:  # pylint: disable=too-few-public-methods
@@ -132,7 +131,7 @@ def _send(self) -> None:
 
     def _send_bytes(
         self,
-        conn: Union["SocketPool.Socket", "socket.socket"],
+        conn: _ISocket,
         buffer: Union[bytes, bytearray, memoryview],
     ):
         bytes_sent: int = 0
@@ -708,7 +707,7 @@ def _read_frame(self):
             length -= min(payload_length, length)
 
         if has_mask:
-            payload = bytes(x ^ mask[i % 4] for i, x in enumerate(payload))
+            payload = bytes(byte ^ mask[idx % 4] for idx, byte in enumerate(payload))
 
         return opcode, payload
 

adafruit_httpserver/server.py

@@ -8,9 +8,7 @@
 """
 
 try:
-    from typing import Callable, Protocol, Union, List, Tuple, Dict, Iterable
-    from socket import socket
-    from socketpool import SocketPool
+    from typing import Callable, Union, List, Tuple, Dict, Iterable
 except ImportError:
     pass
 
@@ -28,6 +26,7 @@
     ServingFilesDisabledError,
 )
 from .headers import Headers
+from .interfaces import _ISocketPool, _ISocket
 from .methods import GET, HEAD
 from .request import Request
 from .response import Response, FileResponse
@@ -54,7 +53,7 @@ class Server:  # pylint: disable=too-many-instance-attributes
     """Root directory to serve files from. ``None`` if serving files is disabled."""
 
     def __init__(
-        self, socket_source: Protocol, root_path: str = None, *, debug: bool = False
+        self, socket_source: _ISocketPool, root_path: str = None, *, debug: bool = False
     ) -> None:
         """Create a server, and get it ready to run.
 
@@ -244,9 +243,7 @@ def stop(self) -> None:
         if self.debug:
             _debug_stopped_server(self)
 
-    def _receive_header_bytes(
-        self, sock: Union["SocketPool.Socket", "socket.socket"]
-    ) -> bytes:
+    def _receive_header_bytes(self, sock: _ISocket) -> bytes:
         """Receive bytes until a empty line is received."""
         received_bytes = bytes()
         while b"\r\n\r\n" not in received_bytes:
@@ -263,7 +260,7 @@ def _receive_header_bytes(
 
     def _receive_body_bytes(
         self,
-        sock: Union["SocketPool.Socket", "socket.socket"],
+        sock: _ISocket,
         received_body_bytes: bytes,
         content_length: int,
     ) -> bytes:
@@ -282,7 +279,7 @@ def _receive_body_bytes(
 
     def _receive_request(
         self,
-        sock: Union["SocketPool.Socket", "socket.socket"],
+        sock: _ISocket,
         client_address: Tuple[str, int],
     ) -> Request:
         """Receive bytes from socket until the whole request is received."""