1  import sys
   2  import types
   3  from socket import SocketType
   4  from typing import Any, BinaryIO, Callable, ClassVar, List, Optional, Text, Tuple, Type, Union
   5  
   6  class BaseServer:
   7      address_family: int
   8      RequestHandlerClass: Callable[..., BaseRequestHandler]
   9      server_address: Tuple[str, int]
  10      socket: SocketType
  11      allow_reuse_address: bool
  12      request_queue_size: int
  13      socket_type: int
  14      timeout: Optional[float]
  15      def __init__(self, server_address: Any, RequestHandlerClass: Callable[..., BaseRequestHandler]) -> None: ...
  16      def fileno(self) -> int: ...
  17      def handle_request(self) -> None: ...
  18      def serve_forever(self, poll_interval: float = ...) -> None: ...
  19      def shutdown(self) -> None: ...
  20      def server_close(self) -> None: ...
  21      def finish_request(self, request: bytes, client_address: Tuple[str, int]) -> None: ...
  22      def get_request(self) -> Tuple[SocketType, Tuple[str, int]]: ...
  23      def handle_error(self, request: bytes, client_address: Tuple[str, int]) -> None: ...
  24      def handle_timeout(self) -> None: ...
  25      def process_request(self, request: bytes, client_address: Tuple[str, int]) -> None: ...
  26      def server_activate(self) -> None: ...
  27      def server_bind(self) -> None: ...
  28      def verify_request(self, request: bytes, client_address: Tuple[str, int]) -> bool: ...
  29  
  30  class TCPServer(BaseServer):
  31      def __init__(
  32          self,
  33          server_address: Tuple[str, int],
  34          RequestHandlerClass: Callable[..., BaseRequestHandler],
  35          bind_and_activate: bool = ...,
  36      ) -> None: ...
  37  
  38  class UDPServer(BaseServer):
  39      def __init__(
  40          self,
  41          server_address: Tuple[str, int],
  42          RequestHandlerClass: Callable[..., BaseRequestHandler],
  43          bind_and_activate: bool = ...,
  44      ) -> None: ...
  45  
  46  if sys.platform != "win32":
  47      class UnixStreamServer(BaseServer):
  48          def __init__(
  49              self,
  50              server_address: Union[Text, bytes],
  51              RequestHandlerClass: Callable[..., BaseRequestHandler],
  52              bind_and_activate: bool = ...,
  53          ) -> None: ...
  54      class UnixDatagramServer(BaseServer):
  55          def __init__(
  56              self,
  57              server_address: Union[Text, bytes],
  58              RequestHandlerClass: Callable[..., BaseRequestHandler],
  59              bind_and_activate: bool = ...,
  60          ) -> None: ...
  61  
  62  if sys.platform != "win32":
  63      class ForkingMixIn:
  64          timeout: Optional[float]  # undocumented
  65          active_children: Optional[List[int]]  # undocumented
  66          max_children: int  # undocumented
  67          def collect_children(self) -> None: ...  # undocumented
  68          def handle_timeout(self) -> None: ...  # undocumented
  69          def process_request(self, request: bytes, client_address: Tuple[str, int]) -> None: ...
  70  
  71  class ThreadingMixIn:
  72      daemon_threads: bool
  73      def process_request_thread(self, request: bytes, client_address: Tuple[str, int]) -> None: ...  # undocumented
  74      def process_request(self, request: bytes, client_address: Tuple[str, int]) -> None: ...
  75  
  76  if sys.platform != "win32":
  77      class ForkingTCPServer(ForkingMixIn, TCPServer): ...
  78      class ForkingUDPServer(ForkingMixIn, UDPServer): ...
  79  
  80  class ThreadingTCPServer(ThreadingMixIn, TCPServer): ...
  81  class ThreadingUDPServer(ThreadingMixIn, UDPServer): ...
  82  
  83  if sys.platform != "win32":
  84      class ThreadingUnixStreamServer(ThreadingMixIn, UnixStreamServer): ...
  85      class ThreadingUnixDatagramServer(ThreadingMixIn, UnixDatagramServer): ...
  86  
  87  class BaseRequestHandler:
  88      # Those are technically of types, respectively:
  89      # * Union[SocketType, Tuple[bytes, SocketType]]
  90      # * Union[Tuple[str, int], str]
  91      # But there are some concerns that having unions here would cause
  92      # too much inconvenience to people using it (see
  93      # https://github.com/python/typeshed/pull/384#issuecomment-234649696)
  94      request: Any
  95      client_address: Any
  96      server: BaseServer
  97      def __init__(self, request: Any, client_address: Any, server: BaseServer) -> None: ...
  98      def setup(self) -> None: ...
  99      def handle(self) -> None: ...
 100      def finish(self) -> None: ...
 101  
 102  class StreamRequestHandler(BaseRequestHandler):
 103      rbufsize: ClassVar[int]  # Undocumented
 104      wbufsize: ClassVar[int]  # Undocumented
 105      timeout: ClassVar[Optional[float]]  # Undocumented
 106      disable_nagle_algorithm: ClassVar[bool]  # Undocumented
 107      connection: SocketType  # Undocumented
 108      rfile: BinaryIO
 109      wfile: BinaryIO
 110  
 111  class DatagramRequestHandler(BaseRequestHandler):
 112      packet: SocketType  # Undocumented
 113      socket: SocketType  # Undocumented
 114      rfile: BinaryIO
 115      wfile: BinaryIO
 116