fix: fix signature inspection on debounced/throttled, update typing and wrapped

pyproject.toml

@@ -115,6 +115,7 @@ exclude = [
     "src/psygnal/containers",
     "src/psygnal/qt.py",
     "src/psygnal/_pyinstaller_util",
+    "src/psygnal/_throttler.py",
 ]
 
 [tool.cibuildwheel]

src/psygnal/_throttler.py

@@ -1,35 +1,53 @@
 from __future__ import annotations
 
 from threading import Timer
-from typing import Any, Callable
+from typing import TYPE_CHECKING, Any, Callable, Generic, TypeVar
 
-from typing_extensions import Literal
+if TYPE_CHECKING:
+    import inspect
 
-Kind = Literal["throttler", "debouncer"]
-EmissionPolicy = Literal["trailing", "leading"]
+    from typing_extensions import Literal, ParamSpec
 
+    Kind = Literal["throttler", "debouncer"]
+    EmissionPolicy = Literal["trailing", "leading"]
 
-class _ThrottlerBase:
+    P = ParamSpec("P")
+else:
+    # just so that we don't have to depend on a new version of typing_extensions
+    # at runtime
+    P = TypeVar("P")
+
+
+class _ThrottlerBase(Generic[P]):
     _timer: Timer
 
     def __init__(
         self,
-        func: Callable[..., Any],
+        func: Callable[P, Any],
         interval: int = 100,
         policy: EmissionPolicy = "leading",
     ) -> None:
-        self._func = func
-        self._interval: int = interval
+        self.__wrapped__: Callable[P, Any] = func
+        self._interval: float = interval / 1000
         self._policy: EmissionPolicy = policy
         self._has_pending: bool = False
         self._timer: Timer = Timer(0, lambda: None)
         self._timer.start()
         self._args: tuple[Any, ...] = ()
         self._kwargs: dict[str, Any] = {}
 
+        # this mimics what functools.wraps does, but avoids __dict__ usage and other
+        # things that won't work with mypyc... HOWEVER, most of these dynamic
+        # assignments won't work in mypyc anyway (they just do nothing.)
+        self.__module__: str = getattr(func, "__module__", "")
+        self.__name__: str = getattr(func, "__name__", "")
+        self.__qualname__: str = getattr(func, "__qualname__", "")
+        self.__doc__: str | None = getattr(func, "__doc__", None)
+        self.__annotations__: dict[str, Any] = getattr(func, "__annotations__", {})
+
     def _actually_call(self) -> None:
         self._has_pending = False
-        self._func(*self._args, **self._kwargs)
+        self.__wrapped__(*self._args, **self._kwargs)
         self._start_timer()
 
     def _call_if_has_pending(self) -> None:
@@ -38,7 +56,7 @@ def _call_if_has_pending(self) -> None:
 
     def _start_timer(self) -> None:
         self._timer.cancel()
-        self._timer = Timer(self._interval / 1000, self._call_if_has_pending)
+        self._timer = Timer(self._interval, self._call_if_has_pending)
         self._timer.start()
 
     def cancel(self) -> None:
@@ -50,16 +68,22 @@ def flush(self) -> None:
         """Force a call if there is one pending."""
         self._call_if_has_pending()
 
-    def __call__(self, *args: Any, **kwargs: Any) -> None:
+    def __call__(self, *args: P.args, **kwargs: P.kwargs) -> None:
         raise NotImplementedError("Subclasses must implement this method.")
 
+    @property
+    def __signature__(self) -> inspect.Signature:
+        import inspect
+
+        return inspect.signature(self.__wrapped__)
+
 
-class Throttler(_ThrottlerBase):
+class Throttler(_ThrottlerBase, Generic[P]):
     """Class that prevents calling `func` more than once per `interval`.
 
     Parameters
     ----------
-    func : Callable[..., Any]
+    func : Callable[P, Any]
         a function to wrap
     interval : int, optional
         the minimum interval in ms that must pass before the function is called again,
@@ -73,13 +97,13 @@ class Throttler(_ThrottlerBase):
 
     def __init__(
         self,
-        func: Callable[..., Any],
+        func: Callable[P, Any],
         interval: int = 100,
         policy: EmissionPolicy = "leading",
     ) -> None:
         super().__init__(func, interval, policy)
 
-    def __call__(self, *args: Any, **kwargs: Any) -> None:
+    def __call__(self, *args: P.args, **kwargs: P.kwargs) -> None:
         """Call underlying function."""
         self._has_pending = True
         self._args = args
@@ -92,12 +116,12 @@ def __call__(self, *args: Any, **kwargs: Any) -> None:
                 self._start_timer()
 
 
-class Debouncer(_ThrottlerBase):
+class Debouncer(_ThrottlerBase, Generic[P]):
     """Class that waits at least `interval` before calling `func`.
 
     Parameters
     ----------
-    func : Callable[..., Any]
+    func : Callable[P, Any]
         a function to wrap
     interval : int, optional
         the minimum interval in ms that must pass before the function is called again,
@@ -111,13 +135,13 @@ class Debouncer(_ThrottlerBase):
 
     def __init__(
         self,
-        func: Callable[..., Any],
+        func: Callable[P, Any],
         interval: int = 100,
         policy: EmissionPolicy = "trailing",
     ) -> None:
         super().__init__(func, interval, policy)
 
-    def __call__(self, *args: Any, **kwargs: Any) -> None:
+    def __call__(self, *args: P.args, **kwargs: P.kwargs) -> None:
         """Call underlying function."""
         self._has_pending = True
         self._args = args
@@ -129,10 +153,10 @@ def __call__(self, *args: Any, **kwargs: Any) -> None:
 
 
 def throttled(
-    func: Callable[..., Any] | None = None,
+    func: Callable[P, Any] | None = None,
     timeout: int = 100,
     leading: bool = True,
-) -> Callable[..., None] | Callable[[Callable[..., Any]], Callable[..., None]]:
+) -> Throttler[P] | Callable[[Callable[P, Any]], Throttler[P]]:
     """Create a throttled function that invokes func at most once per timeout.
 
     The throttled function comes with a `cancel` method to cancel delayed func
@@ -174,18 +198,18 @@ def on_change(val: int)
     ```
     """
 
-    def deco(func: Callable[..., Any]) -> Callable[..., None]:
+    def deco(func: Callable[P, Any]) -> Throttler[P]:
         policy: EmissionPolicy = "leading" if leading else "trailing"
         return Throttler(func, timeout, policy)
 
     return deco(func) if func is not None else deco
 
 
 def debounced(
-    func: Callable[..., Any] | None = None,
+    func: Callable[P, Any] | None = None,
     timeout: int = 100,
     leading: bool = False,
-) -> Callable[..., None] | Callable[[Callable[..., Any]], Callable[..., None]]:
+) -> Debouncer[P] | Callable[[Callable[P, Any]], Debouncer[P]]:
     """Create a debounced function that delays invoking `func`.
 
     `func` will not be invoked until `timeout` ms have elapsed since the last time
@@ -230,7 +254,7 @@ def on_change(val: int)
     ```
     """
 
-    def deco(func: Callable[..., Any]) -> Callable[..., None]:
+    def deco(func: Callable[P, Any]) -> Debouncer[P]:
         policy: EmissionPolicy = "leading" if leading else "trailing"
         return Debouncer(func, timeout, policy)
 

tests/test_throttler.py

@@ -1,7 +1,11 @@
 import time
+from inspect import Parameter, signature
+from typing import Callable
 from unittest.mock import Mock
 
-from psygnal import debounced, throttled
+import pytest
+
+from psygnal import SignalInstance, _compiled, debounced, throttled
 
 
 def test_debounced() -> None:
@@ -79,3 +83,29 @@ def test_flush() -> None:
     f1.flush()
     time.sleep(0.2)
     mock1.assert_called_once()
+
+
+@pytest.mark.parametrize("deco", [debounced, throttled])
+def test_throttled_debounced_signature(deco: Callable) -> None:
+    mock = Mock()
+
+    @deco(timeout=0, leading=True)
+    def f1(x: int) -> None:
+        """Doc."""
+        mock(x)
+
+    # make sure we can still inspect the signature
+    assert signature(f1).parameters["x"] == Parameter(
+        "x", Parameter.POSITIONAL_OR_KEYWORD, annotation=int
+    )
+
+    # make sure these are connectable
+    sig = SignalInstance((int, int, int))
+    sig.connect(f1)
+    sig.emit(1, 2, 3)
+    mock.assert_called_once_with(1)
+
+    if not _compiled:
+        # unfortunately, dynamic assignment of __doc__ and stuff isn't possible in mypyc
+        assert f1.__doc__ == "Doc."
+        assert f1.__name__ == "f1"