Fix BaseServer request and client_address types (#4992)

This commit is contained in:
Michael Lazar
2021-05-28 07:26:39 -04:00
committed by GitHub
parent 99a97195c4
commit dae5817f16

View File

@@ -1,7 +1,7 @@
import sys
import types
from socket import SocketType
from typing import Any, BinaryIO, Callable, ClassVar, List, Optional, Tuple, Type, TypeVar, Union
from typing import Any, BinaryIO, Callable, ClassVar, Optional, Set, Tuple, Type, TypeVar, Union
_T = TypeVar("_T")
@@ -20,19 +20,29 @@ class BaseServer:
def serve_forever(self, poll_interval: float = ...) -> None: ...
def shutdown(self) -> None: ...
def server_close(self) -> None: ...
def finish_request(self, request: bytes, client_address: Tuple[str, int]) -> None: ...
def get_request(self) -> Tuple[SocketType, Tuple[str, int]]: ...
def handle_error(self, request: bytes, client_address: Tuple[str, int]) -> None: ...
def finish_request(
self, request: Union[SocketType, Tuple[bytes, SocketType]], client_address: Union[Tuple[str, int], str]
) -> None: ...
def get_request(self) -> Tuple[Any, Any]: ...
def handle_error(
self, request: Union[SocketType, Tuple[bytes, SocketType]], client_address: Union[Tuple[str, int], str]
) -> None: ...
def handle_timeout(self) -> None: ...
def process_request(self, request: bytes, client_address: Tuple[str, int]) -> None: ...
def process_request(
self, request: Union[SocketType, Tuple[bytes, SocketType]], client_address: Union[Tuple[str, int], str]
) -> None: ...
def server_activate(self) -> None: ...
def server_bind(self) -> None: ...
def verify_request(self, request: bytes, client_address: Tuple[str, int]) -> bool: ...
def verify_request(
self, request: Union[SocketType, Tuple[bytes, SocketType]], client_address: Union[Tuple[str, int], str]
) -> bool: ...
def __enter__(self: _T) -> _T: ...
def __exit__(
self, exc_type: Optional[Type[BaseException]], exc_val: Optional[BaseException], exc_tb: Optional[types.TracebackType]
) -> None: ...
def service_actions(self) -> None: ...
def shutdown_request(self, request: Union[SocketType, Tuple[bytes, SocketType]]) -> None: ... # undocumented
def close_request(self, request: Union[SocketType, Tuple[bytes, SocketType]]) -> None: ... # undocumented
class TCPServer(BaseServer):
def __init__(
@@ -41,6 +51,13 @@ class TCPServer(BaseServer):
RequestHandlerClass: Callable[..., BaseRequestHandler],
bind_and_activate: bool = ...,
) -> None: ...
def get_request(self) -> Tuple[SocketType, Any]: ...
def finish_request(self, request: SocketType, client_address: Union[Tuple[str, int], str]) -> None: ...
def handle_error(self, request: SocketType, client_address: Union[Tuple[str, int], str]) -> None: ...
def process_request(self, request: SocketType, client_address: Union[Tuple[str, int], str]) -> None: ...
def verify_request(self, request: SocketType, client_address: Union[Tuple[str, int], str]) -> bool: ...
def shutdown_request(self, request: SocketType) -> None: ... # undocumented
def close_request(self, request: SocketType) -> None: ... # undocumented
class UDPServer(BaseServer):
def __init__(
@@ -49,6 +66,13 @@ class UDPServer(BaseServer):
RequestHandlerClass: Callable[..., BaseRequestHandler],
bind_and_activate: bool = ...,
) -> None: ...
def get_request(self) -> Tuple[Tuple[bytes, SocketType], Any]: ...
def finish_request(self, request: Tuple[bytes, SocketType], client_address: Union[Tuple[str, int], str]) -> None: ...
def handle_error(self, request: Tuple[bytes, SocketType], client_address: Union[Tuple[str, int], str]) -> None: ...
def process_request(self, request: Tuple[bytes, SocketType], client_address: Union[Tuple[str, int], str]) -> None: ...
def verify_request(self, request: Tuple[bytes, SocketType], client_address: Union[Tuple[str, int], str]) -> bool: ...
def shutdown_request(self, request: Tuple[bytes, SocketType]) -> None: ... # undocumented
def close_request(self, request: Tuple[bytes, SocketType]) -> None: ... # undocumented
if sys.platform != "win32":
class UnixStreamServer(BaseServer):
@@ -69,22 +93,28 @@ if sys.platform != "win32":
if sys.platform != "win32":
class ForkingMixIn:
timeout: Optional[float] # undocumented
active_children: Optional[List[int]] # undocumented
active_children: Optional[Set[int]] # undocumented
max_children: int # undocumented
if sys.version_info >= (3, 7):
block_on_close: bool
def collect_children(self, *, blocking: bool = ...) -> None: ... # undocumented
def handle_timeout(self) -> None: ... # undocumented
def service_actions(self) -> None: ... # undocumented
def process_request(self, request: bytes, client_address: Tuple[str, int]) -> None: ...
def process_request(
self, request: Union[SocketType, Tuple[bytes, SocketType]], client_address: Union[Tuple[str, int], str]
) -> None: ...
def server_close(self) -> None: ...
class ThreadingMixIn:
daemon_threads: bool
if sys.version_info >= (3, 7):
block_on_close: bool
def process_request_thread(self, request: bytes, client_address: Tuple[str, int]) -> None: ... # undocumented
def process_request(self, request: bytes, client_address: Tuple[str, int]) -> None: ...
def process_request_thread(
self, request: Union[SocketType, Tuple[bytes, SocketType]], client_address: Union[Tuple[str, int], str]
) -> None: ... # undocumented
def process_request(
self, request: Union[SocketType, Tuple[bytes, SocketType]], client_address: Union[Tuple[str, int], str]
) -> None: ...
def server_close(self) -> None: ...
if sys.platform != "win32":
@@ -108,7 +138,12 @@ class BaseRequestHandler:
request: Any
client_address: Any
server: BaseServer
def __init__(self, request: Any, client_address: Any, server: BaseServer) -> None: ...
def __init__(
self,
request: Union[SocketType, Tuple[bytes, SocketType]],
client_address: Union[Tuple[str, int], str],
server: BaseServer,
) -> None: ...
def setup(self) -> None: ...
def handle(self) -> None: ...
def finish(self) -> None: ...