gevent: Use TypeVarTuple instead of old ParamSpec workaround. (#11129)

Author: Daverball

stubs/gevent/@tests/stubtest_allowlist.txt

@@ -122,10 +122,6 @@ gevent._config._PositiveValueMixin.validate
 gevent._ffi.watcher.AbstractWatcherType.__new__
 
 # these are inconsistent due to the ParamSpec hack for positional only callables
-gevent._ffi.loop.AbstractLoop.run_callback
-gevent._ffi.loop.AbstractLoop.run_callback_threadsafe
-gevent._ffi.watcher.watcher.start
-gevent._hub_primitives.WaitOperationsGreenlet.cancel_waits_close_and_then
 gevent.baseserver.BaseServer.do_close
 gevent.baseserver.BaseServer.do_handle
 

stubs/gevent/gevent/_ffi/loop.pyi

@@ -3,11 +3,11 @@ from _typeshed import FileDescriptor
 from collections.abc import Callable
 from types import TracebackType
 from typing import Protocol
-from typing_extensions import ParamSpec, TypeAlias
+from typing_extensions import TypeAlias, TypeVarTuple, Unpack
 
 from gevent._types import _AsyncWatcher, _Callback, _ChildWatcher, _IoWatcher, _StatWatcher, _TimerWatcher, _Watcher
 
-_P = ParamSpec("_P")
+_Ts = TypeVarTuple("_Ts")
 _ErrorHandlerFunc: TypeAlias = Callable[
     [object | None, type[BaseException] | None, BaseException | None, TracebackType | None], object
 ]
@@ -74,10 +74,7 @@ class AbstractLoop:
 
     def async_(self, ref: bool = True, priority: int | None = None) -> _AsyncWatcher: ...
     def stat(self, path: str, interval: float = 0.0, ref: bool = True, priority: bool | None = ...) -> _StatWatcher: ...
-    # These technically don't allow the functions arguments to be passed in as kwargs
-    # but there's no way to express that yet with ParamSpec, however, we would still like
-    # to verify that the arguments match
-    def run_callback(self, func: Callable[_P, object], *args: _P.args, **_: _P.kwargs) -> _Callback: ...
-    def run_callback_threadsafe(self, func: Callable[_P, object], *args: _P.args, **_: _P.kwargs) -> _Callback: ...
+    def run_callback(self, func: Callable[[Unpack[_Ts]], object], *args: Unpack[_Ts]) -> _Callback: ...
+    def run_callback_threadsafe(self, func: Callable[[Unpack[_Ts]], object], *args: Unpack[_Ts]) -> _Callback: ...
     def callback(self, priority: float | None = ...) -> _Callback: ...
     def fileno(self) -> FileDescriptor | None: ...

stubs/gevent/gevent/_ffi/watcher.pyi

@@ -2,11 +2,11 @@ from _typeshed import FileDescriptor, StrOrBytesPath
 from collections.abc import Callable
 from types import TracebackType
 from typing import Any, overload
-from typing_extensions import Concatenate, Literal, ParamSpec, Self
+from typing_extensions import Literal, Self, TypeVarTuple, Unpack
 
 from gevent._types import _Loop, _StatResult
 
-_P = ParamSpec("_P")
+_Ts = TypeVarTuple("_Ts")
 
 class AbstractWatcherType(type):
     def new_handle(cls, obj: object) -> int: ...
@@ -22,7 +22,7 @@ class watcher(metaclass=AbstractWatcherType):
     def ref(self) -> bool: ...
     callback: Callable[..., Any]
     args: tuple[Any, ...]
-    def start(self, callback: Callable[_P, Any], *args: _P.args, **_: _P.kwargs) -> None: ...
+    def start(self, callback: Callable[[Unpack[_Ts]], Any], *args: Unpack[_Ts]) -> None: ...
     def stop(self) -> None: ...
     @property
     def priority(self) -> int | None: ...
@@ -36,27 +36,23 @@ class watcher(metaclass=AbstractWatcherType):
 class IoMixin:
     EVENT_MASK: int
     def __init__(self, loop: _Loop, fd: FileDescriptor, events: int, ref: bool = True, priority: int | None = None) -> None: ...
-    # pass_events means the first argument of the callback needs to be an integer, but we can't
-    # type check the other passed in args in this case
     @overload
-    def start(self, callback: Callable[Concatenate[int, _P], Any], *args: Any, pass_events: Literal[True]) -> None: ...
+    def start(self, callback: Callable[[int, Unpack[_Ts]], Any], *args: Unpack[_Ts], pass_events: Literal[True]) -> None: ...
     @overload
-    def start(self, callback: Callable[_P, Any], *args: _P.args, **_: _P.kwargs) -> None: ...
+    def start(self, callback: Callable[[Unpack[_Ts]], Any], *args: Unpack[_Ts]) -> None: ...
 
 class TimerMixin:
     def __init__(
         self, loop: _Loop, after: float = 0.0, repeat: float = 0.0, ref: bool = True, priority: int | None = None
     ) -> None: ...
-    # this has one specific allowed keyword argument, if it is given we don't try to check
-    # the passed in arguments, but if it isn't passed in, then we do.
     @overload
-    def start(self, callback: Callable[..., Any], *args: Any, update: bool) -> None: ...
+    def start(self, callback: Callable[[Unpack[_Ts]], object], *args: Unpack[_Ts], update: bool) -> None: ...
     @overload
-    def start(self, callback: Callable[_P, Any], *args: _P.args, **_: _P.kwargs) -> None: ...
+    def start(self, callback: Callable[[Unpack[_Ts]], object], *args: Unpack[_Ts]) -> None: ...
     @overload
-    def again(self, callback: Callable[..., Any], *args: Any, update: bool) -> None: ...
+    def again(self, callback: Callable[[Unpack[_Ts]], object], *args: Unpack[_Ts], update: bool) -> None: ...
     @overload
-    def again(self, callback: Callable[_P, Any], *args: _P.args, **_: _P.kwargs) -> None: ...
+    def again(self, callback: Callable[[Unpack[_Ts]], object], *args: Unpack[_Ts]) -> None: ...
 
 class SignalMixin:
     def __init__(self, loop: _Loop, signalnum: int, ref: bool = True, priority: int | None = None) -> None: ...

stubs/gevent/gevent/_hub_primitives.pyi

@@ -2,7 +2,7 @@ from _typeshed import FileDescriptor
 from collections.abc import Callable, Collection, Iterable
 from types import TracebackType
 from typing import Any, Generic, Protocol, TypeVar, overload
-from typing_extensions import ParamSpec, Self
+from typing_extensions import Self, TypeVarTuple, Unpack
 
 from gevent._greenlet_primitives import SwitchOutGreenletWithLoop
 from gevent._types import _Loop, _Watcher
@@ -12,8 +12,8 @@ from gevent.socket import socket
 __all__ = ["WaitOperationsGreenlet", "iwait_on_objects", "wait_on_objects", "wait_read", "wait_write", "wait_readwrite"]
 
 _T = TypeVar("_T")
+_Ts = TypeVarTuple("_Ts")
 _WaitableT = TypeVar("_WaitableT", bound=_Waitable)
-_P = ParamSpec("_P")
 
 class _Waitable(Protocol):
     def rawlink(self, __callback: Callable[[Any], object]) -> object: ...
@@ -22,14 +22,12 @@ class _Waitable(Protocol):
 class WaitOperationsGreenlet(SwitchOutGreenletWithLoop):
     loop: _Loop
     def wait(self, watcher: _Watcher) -> None: ...
-    # These then doesn't allow keyword arguments, but ParamSpec doesn't allow for that
     def cancel_waits_close_and_then(
         self,
         watchers: Iterable[_Watcher],
         exc_kind: type[BaseException] | BaseException,
-        then: Callable[_P, object],
-        *then_args: _P.args,
-        **_: _P.kwargs,
+        then: Callable[[Unpack[_Ts]], object],
+        *then_args: Unpack[_Ts],
     ) -> None: ...
     def cancel_wait(self, watcher: _Watcher, error: type[BaseException] | BaseException, close_watcher: bool = False) -> None: ...
 

stubs/gevent/gevent/_types.pyi

@@ -3,9 +3,9 @@ from _typeshed import FileDescriptor, StrOrBytesPath
 from collections.abc import Callable
 from types import TracebackType
 from typing import Any, Protocol, overload
-from typing_extensions import Concatenate, Literal, ParamSpec, TypeAlias
+from typing_extensions import Literal, TypeAlias, TypeVarTuple, Unpack
 
-_P = ParamSpec("_P")
+_Ts = TypeVarTuple("_Ts")
 
 # gevent uses zope.interface interanlly which does not work well with type checkers
 # partially due to the odd call signatures without self and partially due to them
@@ -62,16 +62,13 @@ class _Loop(Protocol):  # noqa: Y046
 
     def async_(self, ref: bool = True, priority: int | None = None) -> _AsyncWatcher: ...
     def stat(self, path: str, interval: float = 0.0, ref: bool = True, priority: bool | None = ...) -> _StatWatcher: ...
-    # These technically don't allow the functions arguments to be passed in as kwargs
-    # but there's no way to express that yet with ParamSpec, however, we would still like
-    # to verify that the arguments match
-    def run_callback(self, func: Callable[_P, Any], *args: _P.args, **_: _P.kwargs) -> _Callback: ...
-    def run_callback_threadsafe(self, func: Callable[_P, Any], *args: _P.args, **_: _P.kwargs) -> _Callback: ...
+    def run_callback(self, func: Callable[[Unpack[_Ts]], Any], *args: Unpack[_Ts]) -> _Callback: ...
+    def run_callback_threadsafe(self, func: Callable[[Unpack[_Ts]], Any], *args: Unpack[_Ts]) -> _Callback: ...
     def fileno(self) -> FileDescriptor | None: ...
 
 class _Watcher(Protocol):
     # while IWatcher allows for kwargs the actual implementation does not...
-    def start(self, callback: Callable[_P, Any], *args: _P.args, **_: _P.kwargs) -> None: ...
+    def start(self, callback: Callable[[Unpack[_Ts]], Any], *args: Unpack[_Ts]) -> None: ...
     def stop(self) -> None: ...
     def close(self) -> None: ...
 
@@ -80,23 +77,23 @@ class _TimerWatcher(_Watcher, Protocol):
     # this has one specific allowed keyword argument, if it is given we don't try to check
     # the passed in arguments, but if it isn't passed in, then we do.
     @overload
-    def start(self, callback: Callable[..., Any], *args: Any, update: bool) -> None: ...
+    def start(self, callback: Callable[[Unpack[_Ts]], Any], *args: Unpack[_Ts], update: bool) -> None: ...
     @overload
-    def start(self, callback: Callable[_P, Any], *args: _P.args, **_: _P.kwargs) -> None: ...
+    def start(self, callback: Callable[[Unpack[_Ts]], Any], *args: Unpack[_Ts]) -> None: ...
     @overload
-    def again(self, callback: Callable[..., Any], *args: Any, update: bool) -> None: ...
+    def again(self, callback: Callable[[Unpack[_Ts]], Any], *args: Unpack[_Ts], update: bool) -> None: ...
     @overload
-    def again(self, callback: Callable[_P, Any], *args: _P.args, **_: _P.kwargs) -> None: ...
+    def again(self, callback: Callable[[Unpack[_Ts]], Any], *args: Unpack[_Ts]) -> None: ...
 
 # this matches Intersection[_Watcher, IoMixin]
 class _IoWatcher(_Watcher, Protocol):
     EVENT_MASK: int
     # pass_events means the first argument of the callback needs to be an integer, but we can't
     # type check the other passed in args in this case
     @overload
-    def start(self, callback: Callable[Concatenate[int, _P], Any], *args: Any, pass_events: Literal[True]) -> None: ...
+    def start(self, callback: Callable[[int, Unpack[_Ts]], Any], *args: Unpack[_Ts], pass_events: Literal[True]) -> None: ...
     @overload
-    def start(self, callback: Callable[_P, Any], *args: _P.args, **_: _P.kwargs) -> None: ...
+    def start(self, callback: Callable[[Unpack[_Ts]], Any], *args: Unpack[_Ts]) -> None: ...
 
 # this matches Intersection[_Watcher, ChildMixin]
 class _ChildWatcher(_Watcher, Protocol):