1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
|
import sys
import types
from _socket import _Address, _RetAddress
from _typeshed import ReadableBuffer
from collections.abc import Callable
from io import BufferedIOBase
from socket import socket as _socket
from typing import Any, ClassVar
from typing_extensions import Self, TypeAlias
__all__ = [
"BaseServer",
"TCPServer",
"UDPServer",
"ThreadingUDPServer",
"ThreadingTCPServer",
"BaseRequestHandler",
"StreamRequestHandler",
"DatagramRequestHandler",
"ThreadingMixIn",
]
if sys.platform != "win32":
__all__ += [
"ForkingMixIn",
"ForkingTCPServer",
"ForkingUDPServer",
"ThreadingUnixDatagramServer",
"ThreadingUnixStreamServer",
"UnixDatagramServer",
"UnixStreamServer",
]
if sys.version_info >= (3, 12):
__all__ += ["ForkingUnixStreamServer", "ForkingUnixDatagramServer"]
_RequestType: TypeAlias = _socket | tuple[bytes, _socket]
_AfUnixAddress: TypeAlias = str | ReadableBuffer # address acceptable for an AF_UNIX socket
_AfInetAddress: TypeAlias = tuple[str | bytes | bytearray, int] # address acceptable for an AF_INET socket
# This can possibly be generic at some point:
class BaseServer:
address_family: int
server_address: _Address
socket: _socket
allow_reuse_address: bool
request_queue_size: int
socket_type: int
timeout: float | None
RequestHandlerClass: Callable[[Any, _RetAddress, Self], BaseRequestHandler]
def __init__(
self, server_address: _Address, RequestHandlerClass: Callable[[Any, _RetAddress, Self], BaseRequestHandler]
) -> None: ...
def fileno(self) -> int: ...
def handle_request(self) -> None: ...
def serve_forever(self, poll_interval: float = 0.5) -> None: ...
def shutdown(self) -> None: ...
def server_close(self) -> None: ...
def finish_request(self, request: _RequestType, client_address: _RetAddress) -> None: ...
def get_request(self) -> tuple[Any, Any]: ...
def handle_error(self, request: _RequestType, client_address: _RetAddress) -> None: ...
def handle_timeout(self) -> None: ...
def process_request(self, request: _RequestType, client_address: _RetAddress) -> None: ...
def server_activate(self) -> None: ...
def server_bind(self) -> None: ...
def verify_request(self, request: _RequestType, client_address: _RetAddress) -> bool: ...
def __enter__(self) -> Self: ...
def __exit__(
self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: types.TracebackType | None
) -> None: ...
def service_actions(self) -> None: ...
def shutdown_request(self, request: _RequestType) -> None: ... # undocumented
def close_request(self, request: _RequestType) -> None: ... # undocumented
class TCPServer(BaseServer):
if sys.version_info >= (3, 11):
allow_reuse_port: bool
server_address: _AfInetAddress
def __init__(
self,
server_address: _AfInetAddress,
RequestHandlerClass: Callable[[Any, _RetAddress, Self], BaseRequestHandler],
bind_and_activate: bool = True,
) -> None: ...
def get_request(self) -> tuple[_socket, _RetAddress]: ...
class UDPServer(TCPServer):
max_packet_size: ClassVar[int]
def get_request(self) -> tuple[tuple[bytes, _socket], _RetAddress]: ... # type: ignore[override]
if sys.platform != "win32":
class UnixStreamServer(TCPServer):
server_address: _AfUnixAddress # type: ignore[assignment]
def __init__(
self,
server_address: _AfUnixAddress,
RequestHandlerClass: Callable[[Any, _RetAddress, Self], BaseRequestHandler],
bind_and_activate: bool = True,
) -> None: ...
class UnixDatagramServer(UDPServer):
server_address: _AfUnixAddress # type: ignore[assignment]
def __init__(
self,
server_address: _AfUnixAddress,
RequestHandlerClass: Callable[[Any, _RetAddress, Self], BaseRequestHandler],
bind_and_activate: bool = True,
) -> None: ...
if sys.platform != "win32":
class ForkingMixIn:
timeout: float | None # undocumented
active_children: set[int] | None # undocumented
max_children: int # undocumented
block_on_close: bool
def collect_children(self, *, blocking: bool = False) -> None: ... # undocumented
def handle_timeout(self) -> None: ... # undocumented
def service_actions(self) -> None: ... # undocumented
def process_request(self, request: _RequestType, client_address: _RetAddress) -> None: ...
def server_close(self) -> None: ...
class ThreadingMixIn:
daemon_threads: bool
block_on_close: bool
def process_request_thread(self, request: _RequestType, client_address: _RetAddress) -> None: ... # undocumented
def process_request(self, request: _RequestType, client_address: _RetAddress) -> None: ...
def server_close(self) -> None: ...
if sys.platform != "win32":
class ForkingTCPServer(ForkingMixIn, TCPServer): ...
class ForkingUDPServer(ForkingMixIn, UDPServer): ...
if sys.version_info >= (3, 12):
class ForkingUnixStreamServer(ForkingMixIn, UnixStreamServer): ...
class ForkingUnixDatagramServer(ForkingMixIn, UnixDatagramServer): ...
class ThreadingTCPServer(ThreadingMixIn, TCPServer): ...
class ThreadingUDPServer(ThreadingMixIn, UDPServer): ...
if sys.platform != "win32":
class ThreadingUnixStreamServer(ThreadingMixIn, UnixStreamServer): ...
class ThreadingUnixDatagramServer(ThreadingMixIn, UnixDatagramServer): ...
class BaseRequestHandler:
# `request` is technically of type _RequestType,
# but there are some concerns that having a union here would cause
# too much inconvenience to people using it (see
# https://github.com/python/typeshed/pull/384#issuecomment-234649696)
#
# Note also that _RetAddress is also just an alias for `Any`
request: Any
client_address: _RetAddress
server: BaseServer
def __init__(self, request: _RequestType, client_address: _RetAddress, server: BaseServer) -> None: ...
def setup(self) -> None: ...
def handle(self) -> None: ...
def finish(self) -> None: ...
class StreamRequestHandler(BaseRequestHandler):
rbufsize: ClassVar[int] # undocumented
wbufsize: ClassVar[int] # undocumented
timeout: ClassVar[float | None] # undocumented
disable_nagle_algorithm: ClassVar[bool] # undocumented
connection: Any # undocumented
rfile: BufferedIOBase
wfile: BufferedIOBase
class DatagramRequestHandler(BaseRequestHandler):
packet: bytes # undocumented
socket: _socket # undocumented
rfile: BufferedIOBase
wfile: BufferedIOBase
|
