Use Callable for typing, rather than ParamSpec (#710)

Author: Dreamsorcerer

.coveragerc

@@ -9,14 +9,6 @@ _site-packages-to-src-mapping =
   */lib/python*/site-packages
   *\Lib\site-packages
 
-[report]
-fail_under = 98
-skip_covered = true
-skip_empty = true
-show_missing = true
-exclude_also =
-  ^\s*@pytest\.mark\.xfail
-
 [run]
 branch = true
 cover_pylib = false

CHANGES/699.feature.rst

@@ -1,4 +1,2 @@
-Improved type safety by allowing callback parameters to be type checked (``typing-extensions`` is now required for Python 3.9).
-
 Added decorator functionality to ``Signal`` as a convenient way to add a callback
 -- by ``@Vizonex``.

CHANGES/710.feature.rst

@@ -0,0 +1,2 @@
+Improved type safety by allowing callback parameters to be type checked (``typing-extensions`` is now required for Python <3.13).
+-- by ``@Vizonex`` and ``@Dreamsorcerer``.

aiosignal/__init__.py

@@ -1,23 +1,27 @@
 import sys
-from typing import Awaitable, Callable, TypeVar
+from typing import Any, Awaitable, Callable, TypeVar
 
 from frozenlist import FrozenList
 
-if sys.version_info >= (3, 10):
-    from typing import ParamSpec
+if sys.version_info >= (3, 11):
+    from typing import Unpack
 else:
-    from typing_extensions import ParamSpec
+    from typing_extensions import Unpack
+
+if sys.version_info >= (3, 13):
+    from typing import TypeVarTuple
+else:
+    from typing_extensions import TypeVarTuple
 
-_P = ParamSpec("_P")
 _T = TypeVar("_T")
-_AsyncFunc = Callable[_P, Awaitable[_T]]
+_Ts = TypeVarTuple("_Ts", default=Unpack[tuple[()]])
 
 __version__ = "1.3.2"
 
 __all__ = ("Signal",)
 
 
-class Signal(FrozenList[_AsyncFunc[_P, _T]]):
+class Signal(FrozenList[Callable[[Unpack[_Ts]], Awaitable[object]]]):
     """Coroutine-based signal implementation.
 
     To connect a callback to a signal, use any list method.
@@ -37,7 +41,7 @@ def __repr__(self) -> str:
             self._owner, self.frozen, list(self)
         )
 
-    async def send(self, *args: _P.args, **kwargs: _P.kwargs) -> None:
+    async def send(self, *args: Unpack[_Ts], **kwargs: Any) -> None:
         """
         Sends data to all registered receivers.
         """
@@ -47,7 +51,9 @@ async def send(self, *args: _P.args, **kwargs: _P.kwargs) -> None:
         for receiver in self:
             await receiver(*args, **kwargs)
 
-    def __call__(self, func: _AsyncFunc[_P, _T]) -> _AsyncFunc[_P, _T]:
+    def __call__(
+        self, func: Callable[[Unpack[_Ts]], Awaitable[_T]]
+    ) -> Callable[[Unpack[_Ts]], Awaitable[_T]]:
         """Decorator to add a function to this Signal."""
         self.append(func)
         return func

docs/index.rst

@@ -15,6 +15,14 @@ and dropping callbacks is forbidden.
 The only available operation is calling the previously registered
 callbacks by using ``await sig.send(data)``.
 
+The callback parameters, which should be passed in the ``.send()`` call, can be
+specified for a type checker:
+
+```python
+signal = Signal[int, str](owner)
+signal.send(42, "foo")
+```
+
 For concrete usage examples see the :ref:`aiohttp:aiohttp-web-signals` section of the :doc:`aiohttp:web_advanced` chapter of the :doc:`aiohttp documentation <aiohttp:index>`.
 
 API

setup.cfg

@@ -38,7 +38,7 @@ include_package_data = True
 
 install_requires =
   frozenlist >= 1.1.0
-  typing-extensions >= 3.10; python_version <= '3.9'
+  typing-extensions >= 4.2; python_version < '3.13'
 
 
 [pep8]