gevent: Use a TypeVarTuple in gevent.baseserver.BaseServer

stubs/gevent/@tests/stubtest_allowlist.txt

@@ -120,10 +120,6 @@ gevent._config._PositiveValueMixin.validate
 # internal API implementation detail we don't care about
 gevent._ffi.watcher.AbstractWatcherType.__new__
 
-# these are inconsistent due to the ParamSpec hack for positional only callables
-gevent.baseserver.BaseServer.do_close
-gevent.baseserver.BaseServer.do_handle
-
 # we don't care about write/writeall allowing a named parameter
 gevent._fileobjectcommon.FlushingBufferedWriter.write
 gevent._fileobjectcommon.WriteIsWriteallMixin.write

stubs/gevent/gevent/baseserver.pyi

@@ -1,21 +1,22 @@
 from collections.abc import Callable, Container
 from types import TracebackType
-from typing import Any, Generic, Literal, Protocol
-from typing_extensions import ParamSpec, Self, TypeAlias
+from typing import Generic, Literal, Protocol
+from typing_extensions import ParamSpec, Self, TypeAlias, TypeVarTuple, Unpack
 
 from gevent._types import _Loop
 from gevent.pool import Pool
 from gevent.socket import socket as _GeventSocket
 from greenlet import greenlet
 
+_Ts = TypeVarTuple("_Ts")
 _P = ParamSpec("_P")
 
 class _SpawnFunc(Protocol):
     def __call__(self, func: Callable[_P, object], /, *args: _P.args, **kwargs: _P.kwargs) -> greenlet: ...
 
 _Spawner: TypeAlias = Pool | _SpawnFunc | int | Literal["default"] | None
 
-class BaseServer(Generic[_P]):
+class BaseServer(Generic[Unpack[_Ts]]):
     min_delay: float
     max_delay: float
     max_accept: int
@@ -27,27 +28,23 @@ class BaseServer(Generic[_P]):
     family: int
     address: str | tuple[str, int]
     socket: _GeventSocket
-    handle: Callable[..., object]
+    handle: Callable[[Unpack[_Ts]], object]
     def __init__(
         self,
         listener: _GeventSocket | tuple[str, int] | str,
-        handle: Callable[_P, object] | None = None,
+        handle: Callable[[Unpack[_Ts]], object] | None = None,
         spawn: _Spawner = "default",
     ) -> None: ...
     def __enter__(self) -> Self: ...
     def __exit__(self, typ: type[BaseException] | None, value: BaseException | None, tb: TracebackType | None, /) -> None: ...
     def set_listener(self, listener: _GeventSocket | tuple[str, int] | str) -> None: ...
     def set_spawn(self, spawn: _Spawner) -> None: ...
-    def set_handle(self, handle: Callable[_P, object]) -> None: ...
+    def set_handle(self, handle: Callable[[Unpack[_Ts]], object]) -> None: ...
     def start_accepting(self) -> None: ...
     def stop_accepting(self) -> None: ...
-    # neither of these accept keyword arguments, but if we omit them, then ParamSpec
-    # won't match the arguments correctly
-    def do_handle(self, *args: _P.args, **_: _P.kwargs) -> None: ...
-    def do_close(self, *args: _P.args, **_: _P.kwargs) -> None: ...
-    # we would like to return _P.args here, however pyright will complain
-    # mypy doesn't seem to mind
-    def do_read(self) -> tuple[Any, ...] | None: ...
+    def do_handle(self, *args: Unpack[_Ts]) -> None: ...
+    def do_close(self, *args: Unpack[_Ts]) -> None: ...
+    def do_read(self) -> tuple[Unpack[_Ts]] | None: ...
     def full(self) -> bool: ...
     @property
     def server_host(self) -> str | None: ...

stubs/gevent/gevent/server.pyi

@@ -23,7 +23,7 @@ class _SSLArguments(TypedDict, total=False):
     do_handshake_on_connect: bool
     ciphers: str
 
-class StreamServer(BaseServer[[_GeventSocket, _Address]]):
+class StreamServer(BaseServer[_GeventSocket, _Address]):
     backlog: int
     reuse_addr: ClassVar[int | None]
     wrap_socket = ssl_wrap_socket
@@ -67,7 +67,7 @@ class StreamServer(BaseServer[[_GeventSocket, _Address]]):
     def do_close(self, sock: _GeventSocket, address: _Address) -> None: ...
     def wrap_socket_and_handle(self, client_socket: _GeventSocket, address: _StrictAddress) -> Any: ...
 
-class DatagramServer(BaseServer[[_GeventSocket, _Address]]):
+class DatagramServer(BaseServer[_GeventSocket, _Address]):
     reuse_addr: ClassVar[int | None]
     def __init__(
         self,