gunicorn, draft

Author: pajod

stubs/gunicorn/gunicorn/_types.pyi

@@ -1,14 +1,18 @@
 ### This .pyi file is a helper for centralized storage types that are reused across different runtime modules. ###
-from _typeshed import FileDescriptor
+from _typeshed import FileDescriptor, OptExcInfo
 from collections.abc import Awaitable, Callable, Iterable, MutableMapping
-from typing import Any
+from typing import Any, Protocol
 from typing_extensions import LiteralString, TypeAlias
 
 _StatusType: TypeAlias = str
 _HeadersType: TypeAlias = Iterable[tuple[str, str]]
 
 _EnvironType: TypeAlias = MutableMapping[str, Any]  # See https://peps.python.org/pep-0333/
-_StartResponseType: TypeAlias = Callable[[_StatusType, _HeadersType], None]
+class _StartResponseType(Protocol):
+    def __call__(
+        self, status: _StatusType, headers: _HeadersType, exc_info: OptExcInfo | None = ..., /
+    ) -> Callable[[bytes], object]: ...
+
 _ResponseBodyType: TypeAlias = Iterable[bytes]
 _WSGIAppType: TypeAlias = Callable[[_EnvironType, _StartResponseType], _ResponseBodyType]  # noqa: Y047
 

stubs/gunicorn/gunicorn/debug.pyi

@@ -11,7 +11,7 @@ class Spew:
 
     def __init__(self, trace_names: Container[str] | None = None, show_values: bool = True) -> None: ...
     def __call__(
-        self, frame: FrameType, event: str, arg: Any  # `arg` is not used inside the function, stub is set Any
+        self, frame: FrameType, event: Literal["call", "line", "return", "exception", "opcode"], arg: Any  # `arg` is not used inside the function, stub is set Any
     ) -> Self: ...
 
 def spew(trace_names: Container[str] | None = None, show_values: bool = False) -> None: ...

stubs/gunicorn/gunicorn/http/wsgi.pyi

@@ -4,7 +4,7 @@ import re
 import socket
 from _typeshed import ReadableBuffer
 from collections.abc import Callable
-from typing import Any
+from typing import Any, Never
 
 from gunicorn.config import Config
 from gunicorn.http import Request
@@ -21,7 +21,7 @@ class FileWrapper:
     close: Callable[[], None] | None
 
     def __init__(self, filelike: io.IOBase, blksize: int = 8192) -> None: ...
-    def __getitem__(self, key: Any) -> bytes: ...
+    def __getitem__(self, key: Never) -> bytes: ...
 
 class WSGIErrorsWrapper(io.RawIOBase):
     streams: list[io.TextIOBase]

stubs/gunicorn/gunicorn/util.pyi

@@ -12,8 +12,8 @@ hop_headers: set[str]
 
 def load_entry_point(distribution: str, group: str, name: str) -> type[object]: ...
 def load_class(
-    uri: str | object, default: str = "gunicorn.workers.sync.SyncWorker", section: str = "gunicorn.workers"
-) -> type[Any]: ...
+    uri: str | type[Logger] | type[SyncWorker], default: str = "gunicorn.workers.sync.SyncWorker", section: str = "gunicorn.workers"
+) -> type[Logger] | type[SyncWorker]: ...
 
 positionals: tuple[Literal[_ParameterKind.POSITIONAL_ONLY], Literal[_ParameterKind.POSITIONAL_OR_KEYWORD]]
 
@@ -30,19 +30,19 @@ def close(sock: socket) -> None: ...
 def write_chunk(sock: socket, data: bytes) -> None: ...
 def write(sock: socket, data: bytes, chunked: bool = False) -> None: ...
 def write_nonblock(sock: socket, data: bytes, chunked: bool = False) -> None: ...
-def write_error(sock: socket, status_int: int, reason: str, mesg: str) -> None: ...
+def write_error(sock: socket, status_int: int, reason: LiteralString, mesg: str) -> None: ...
 def import_app(module: str) -> _WSGIAppType: ...
 def getcwd() -> str: ...
 def http_date(timestamp: float | None = None) -> str: ...
 def is_hoppish(header: str) -> bool: ...
 def daemonize(enable_stdio_inheritance: bool = False) -> None: ...
 def seed() -> None: ...
 def check_is_writable(path: FileDescriptorOrPath) -> None: ...
-def to_bytestring(value: str | bytes, encoding: str = "utf8") -> bytes: ...
+def to_bytestring(value: str | bytes, encoding: LiteralString = "utf8") -> bytes: ...
 def has_fileno(obj: HasFileno) -> bool: ...
 def warn(msg: str) -> None: ...
 def make_fail_app(msg: str) -> _WSGIAppType: ...
 def split_request_uri(uri: str) -> SplitResult: ...
 def reraise(tp: type[BaseException], value: BaseException | None, tb: types.TracebackType | None = None) -> NoReturn: ...
-def bytes_to_str(b: bytes) -> str: ...
+def bytes_to_str(b: bytes | str) -> str: ...
 def unquote_to_wsgi_str(string: str) -> str: ...

stubs/gunicorn/gunicorn/workers/base.pyi

@@ -19,7 +19,7 @@ class Worker:
     ppid: int
     sockets: list[socket.socket]
     app: BaseApplication
-    timeout: int
+    timeout: float
     cfg: Config
     booted: bool
     aborted: bool
@@ -33,7 +33,7 @@ class Worker:
     wsgi: _WSGIAppType
 
     def __init__(
-        self, age: int, ppid: int, sockets: list[socket.socket], app: BaseApplication, timeout: int, cfg: Config, log: GLogger
+        self, age: int, ppid: int, sockets: list[socket.socket], app: BaseApplication, timeout: float, cfg: Config, log: GLogger
     ) -> None: ...
     def notify(self) -> None: ...
     def run(self) -> None: ...

stubs/gunicorn/gunicorn/workers/base_async.pyi

@@ -1,4 +1,5 @@
 import socket
+from contextlib import AbstractAsyncContextManager
 
 from gunicorn.http import Request
 from gunicorn.workers import base
@@ -11,7 +12,7 @@ class AsyncWorker(base.Worker):
     worker_connections: int
     alive: bool
 
-    def timeout_ctx(self) -> None: ...
+    def timeout_ctx(self) -> AbstractAsyncContextManager[None]: ...
     def is_already_handled(self, respiter: object) -> bool: ...
     def handle(self, listener: socket.socket, client: socket.socket, addr: _AddressType) -> None: ...
-    def handle_request(self, listener_name: str, req: Request, sock: socket.socket, addr: _AddressType) -> bool: ...
+    def handle_request(self, listener_name: str, req: Request, sock: socket.socket, addr: _AddressType) -> bool | None: ...

stubs/gunicorn/gunicorn/workers/geventlet.pyi

@@ -1,3 +1,4 @@
+from contextlib import AbstractAsyncContextManager
 from types import FrameType
 from typing import Any
 from typing_extensions import TypeAlias
@@ -19,6 +20,6 @@ class EventletWorker(AsyncWorker):
     def init_process(self) -> None: ...
     def handle_quit(self, sig: int, frame: FrameType | None) -> None: ...
     def handle_usr1(self, sig: int, frame: FrameType | None) -> None: ...
-    def timeout_ctx(self) -> None: ...
+    def timeout_ctx(self) -> AbstractAsyncContextManager[None]: ...
     def handle(self, listener: GreenSocket, client: GreenSocket, addr: _AddressType) -> None: ...
     def run(self) -> None: ...

stubs/gunicorn/gunicorn/workers/ggevent.pyi

@@ -1,3 +1,5 @@
+from contextlib import AbstractAsyncContextManager
+
 from types import FrameType
 from typing import Any, ClassVar
 
@@ -19,10 +21,10 @@ class GeventWorker(AsyncWorker):
 
     def patch(self) -> None: ...
     def notify(self) -> None: ...
-    def timeout_ctx(self) -> None: ...
+    def timeout_ctx(self) -> AbstractAsyncContextManager[None]: ...
     def run(self) -> None: ...
     def handle(self, listener: GeventSocket, client: GeventSocket, addr: _AddressType) -> None: ...
-    def handle_request(self, listener_name: str, req: Request, sock: GeventSocket, addr: _AddressType) -> bool: ...
+    def handle_request(self, listener_name: _AddressType, req: Request, sock: GeventSocket, addr: _AddressType) -> bool: ...
     def handle_quit(self, sig: int, frame: FrameType | None) -> None: ...
     def handle_usr1(self, sig: int, frame: FrameType | None) -> None: ...
     def init_process(self) -> None: ...

stubs/gunicorn/gunicorn/workers/gthread.pyi

@@ -1,7 +1,7 @@
 import socket
 from collections import deque
 from concurrent.futures import Future, ThreadPoolExecutor
-from selectors import DefaultSelector
+from selectors import BaseSelector
 from types import FrameType
 
 from gunicorn.config import Config
@@ -29,7 +29,7 @@ class ThreadWorker(base.Worker):
     worker_connections: int
     max_keepalived: int
     tpool: ThreadPoolExecutor
-    poller: DefaultSelector
+    poller: BaseSelector
     futures: deque[Future[tuple[bool, TConn]]]
     nr_conns: int
     alive: bool

stubs/gunicorn/gunicorn/workers/sync.pyi

@@ -9,10 +9,10 @@ class StopWaiting(Exception): ...
 
 class SyncWorker(Worker):
     def accept(self, listener: socket.socket) -> None: ...
-    def wait(self, timeout: int) -> list[socket.socket | int] | None: ...
+    def wait(self, timeout: float) -> list[socket.socket]: ...
     def is_parent_alive(self) -> bool: ...
-    def run_for_one(self, timeout: int) -> None: ...
-    def run_for_multiple(self, timeout: int) -> None: ...
+    def run_for_one(self, timeout: float) -> None: ...
+    def run_for_multiple(self, timeout: float) -> None: ...
     def run(self) -> None: ...
     def handle(self, listener: socket.socket, client: socket.socket, addr: _AddressType) -> None: ...
     def handle_request(self, listener: socket.socket, req: Request, client: socket.socket, addr: tuple[str, int]) -> bool: ...