Add stubs for gunicorn (#14690)

Author: eugeneliukindev

pyrightconfig.stricter.json

@@ -44,6 +44,7 @@
         "stubs/grpcio-reflection/grpc_reflection/v1alpha",
         "stubs/grpcio-status/grpc_status",
         "stubs/grpcio/grpc/__init__.pyi",
+        "stubs/gunicorn",
         "stubs/hdbcli/hdbcli/dbapi.pyi",
         "stubs/html5lib",
         "stubs/httplib2",

stubs/gunicorn/@tests/stubtest_allowlist.txt

@@ -0,0 +1,5 @@
+# Is not present in runtime. For more details, see the file itself.
+gunicorn._types
+
+# .pyi file doesn't exist
+gunicorn.__main__

stubs/gunicorn/METADATA.toml

@@ -0,0 +1,8 @@
+version = "23.0.*"
+upstream_repository = "https://github.com/benoitc/gunicorn"
+requires = ["types-gevent"]
+
+[tool.stubtest]
+supported_platforms = ["linux", "darwin"]
+ci_platforms = ["linux", "darwin"]
+stubtest_requirements = ["gevent>=1.4.0", "eventlet>=0.24.1,!=0.36.0", "tornado>=0.2", "setproctitle", "PasteDeploy", "inotify"]

stubs/gunicorn/gunicorn/__init__.pyi

@@ -0,0 +1,4 @@
+version_info: tuple[int, int, int]
+__version__: str
+SERVER: str
+SERVER_SOFTWARE: str

stubs/gunicorn/gunicorn/_types.pyi

@@ -0,0 +1,23 @@
+### This .pyi file is a helper for centralized storage types that are reused across different runtime modules. ###
+from _typeshed import FileDescriptor
+from collections.abc import Awaitable, Callable, Iterable, MutableMapping
+from typing import Any
+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]
+_ResponseBodyType: TypeAlias = Iterable[bytes]
+_WSGIAppType: TypeAlias = Callable[[_EnvironType, _StartResponseType], _ResponseBodyType]  # noqa: Y047
+
+_ScopeType: TypeAlias = MutableMapping[str, Any]
+_MessageType: TypeAlias = MutableMapping[str, Any]
+_ReceiveType: TypeAlias = Callable[[], Awaitable[_MessageType]]
+_SendType: TypeAlias = Callable[[_MessageType], Awaitable[None]]
+_ASGIAppType: TypeAlias = Callable[[_ScopeType, _ReceiveType, _SendType], Awaitable[None]]  # noqa: Y047
+
+_UnixSocketPathType: TypeAlias = str
+_TcpAddressType: TypeAlias = tuple[LiteralString, int]  # noqa: Y047
+_AddressType: TypeAlias = _UnixSocketPathType | FileDescriptor | _TcpAddressType  # noqa: Y047

stubs/gunicorn/gunicorn/app/__init__.pyi

@@ -0,0 +1,36 @@
+from argparse import ArgumentParser, Namespace
+from typing import Any
+from typing_extensions import override
+
+from gunicorn.config import Config
+from gunicorn.glogging import Logger as GLogger
+
+from .._types import _ASGIAppType, _WSGIAppType
+
+class BaseApplication:
+    usage: str | None
+    cfg: Config
+    callable: _WSGIAppType | _ASGIAppType | None
+    prog: str | None
+    logger: GLogger
+
+    def __init__(self, usage: str | None = None, prog: str | None = None) -> None: ...
+    def do_load_config(self) -> None: ...
+    def load_default_config(self) -> None: ...
+    def init(self, parser: ArgumentParser, opts: Namespace, args: list[str]) -> dict[str, Any] | None: ...
+    def load(self) -> _WSGIAppType | _ASGIAppType: ...
+    def load_config(self) -> None: ...
+    def reload(self) -> None: ...
+    def wsgi(self) -> _WSGIAppType | _ASGIAppType: ...
+    def run(self) -> None: ...
+
+class Application(BaseApplication):
+    def chdir(self) -> None: ...
+    def get_config_from_filename(self, filename: str) -> dict[str, Any]: ...
+    def get_config_from_module_name(self, module_name: str) -> dict[str, Any]: ...
+    def load_config_from_module_name_or_filename(self, location: str) -> dict[str, Any]: ...
+    def load_config_from_file(self, filename: str) -> dict[str, Any]: ...
+    @override
+    def load_config(self) -> None: ...
+    @override
+    def run(self) -> None: ...

stubs/gunicorn/gunicorn/app/base.pyi

@@ -0,0 +1,7 @@
+from typing import Any
+
+from .._types import _WSGIAppType
+
+def get_wsgi_app(config_uri: str, name: str | None = None, defaults: dict[str, Any] | None = None) -> _WSGIAppType: ...
+def has_logging_config(config_file: str) -> bool: ...
+def serve(app: _WSGIAppType, global_conf: dict[str, Any], **local_conf: Any) -> None: ...

stubs/gunicorn/gunicorn/app/pasterapp.pyi

@@ -0,0 +1,20 @@
+from argparse import ArgumentParser, Namespace
+from typing_extensions import override
+
+from gunicorn.app.base import Application
+
+from .._types import _WSGIAppType
+
+class WSGIApplication(Application):
+    app_uri: str | None
+
+    @override
+    def init(self, parser: ArgumentParser, opts: Namespace, args: list[str]) -> None: ...
+    @override
+    def load_config(self) -> None: ...
+    def load_wsgiapp(self) -> _WSGIAppType: ...
+    def load_pasteapp(self) -> _WSGIAppType: ...
+    @override
+    def load(self) -> _WSGIAppType: ...
+
+def run(prog: str | None = None) -> None: ...

stubs/gunicorn/gunicorn/app/wsgiapp.pyi

@@ -0,0 +1,68 @@
+import socket
+from types import FrameType
+from typing import ClassVar
+
+from gunicorn.app.base import BaseApplication
+from gunicorn.config import Config
+from gunicorn.glogging import Logger as GLogger
+from gunicorn.workers.base import Worker
+
+from ._types import _AddressType
+from .pidfile import Pidfile
+
+class Arbiter:
+    WORKER_BOOT_ERROR: ClassVar[int]
+    APP_LOAD_ERROR: ClassVar[int]
+    START_CTX: ClassVar[dict[int | str, str | list[str]]]
+    LISTENERS: ClassVar[list[socket.socket]]
+    WORKERS: ClassVar[dict[int, Worker]]
+    PIPE: ClassVar[list[int]]
+    SIG_QUEUE: ClassVar[list[int]]
+    SIGNALS: ClassVar[list[int]]
+    SIG_NAMES: ClassVar[dict[int, str]]
+    log: GLogger | None
+    pidfile: Pidfile | None
+    systemd: bool
+    worker_age: int
+    reexec_pid: int
+    master_pid: int
+    master_name: str
+    pid: int
+    app: BaseApplication
+    cfg: Config
+    worker_class: type[Worker]
+    address: list[_AddressType]
+    timeout: int
+    proc_name: str
+    num_workers: int
+
+    def __init__(self, app: BaseApplication) -> None: ...
+    def setup(self, app: BaseApplication) -> None: ...
+    def start(self) -> None: ...
+    def init_signals(self) -> None: ...
+    def signal(self, sig: int, frame: FrameType | None) -> None: ...
+    def run(self) -> None: ...
+    def handle_chld(self, sig: int, frame: FrameType | None) -> None: ...
+    def handle_hup(self) -> None: ...
+    def handle_term(self) -> None: ...
+    def handle_int(self) -> None: ...
+    def handle_quit(self) -> None: ...
+    def handle_ttin(self) -> None: ...
+    def handle_ttou(self) -> None: ...
+    def handle_usr1(self) -> None: ...
+    def handle_usr2(self) -> None: ...
+    def handle_winch(self) -> None: ...
+    def maybe_promote_master(self) -> None: ...
+    def wakeup(self) -> None: ...
+    def halt(self, reason: str | None = None, exit_status: int = 0) -> None: ...
+    def sleep(self) -> None: ...
+    def stop(self, graceful: bool = True) -> None: ...
+    def reexec(self) -> None: ...
+    def reload(self) -> None: ...
+    def murder_workers(self) -> None: ...
+    def reap_workers(self) -> None: ...
+    def manage_workers(self) -> None: ...
+    def spawn_worker(self) -> int: ...
+    def spawn_workers(self) -> None: ...
+    def kill_workers(self, sig: int) -> None: ...
+    def kill_worker(self, pid: int, sig: int) -> None: ...