use collections.abc instead of typing

CHANGES.rst

@@ -4,7 +4,8 @@ Version 1.8.1
 Unreleased
 
 -   Restore identity handling for ``str`` and ``int`` senders. :pr:`148`
--   Fix deprecated `blinker.base.WeakNamespace` import. :pr:`149`
+-   Fix deprecated ``blinker.base.WeakNamespace`` import. :pr:`149`
+-   Use types from ``collections.abc`` instead of ``typing``. :pr:`150`
 
 
 Version 1.8.0

src/blinker/_utilities.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import collections.abc as c
 import inspect
 import typing as t
 from weakref import ref
@@ -34,11 +35,11 @@ def __init__(self, name: str) -> None:
     def __repr__(self) -> str:
         return self.name
 
-    def __getnewargs__(self) -> tuple[t.Any]:
+    def __getnewargs__(self) -> tuple[t.Any, ...]:
         return (self.name,)
 
 
-def make_id(obj: object) -> t.Hashable:
+def make_id(obj: object) -> c.Hashable:
     """Get a stable identifier for a receiver or sender, to be used as a dict
     key or in a set.
     """
@@ -56,7 +57,7 @@ def make_id(obj: object) -> t.Hashable:
     return id(obj)
 
 
-def make_ref(obj: T, callback: t.Callable[[ref[T]], None] | None = None) -> ref[T]:
+def make_ref(obj: T, callback: c.Callable[[ref[T]], None] | None = None) -> ref[T]:
     if inspect.ismethod(obj):
         return WeakMethod(obj, callback)  # type: ignore[arg-type, return-value]
 

src/blinker/base.py

@@ -9,10 +9,12 @@
 
 from __future__ import annotations
 
+import collections.abc as c
 import typing as t
 import warnings
 import weakref
 from collections import defaultdict
+from contextlib import AbstractContextManager
 from contextlib import contextmanager
 from functools import cached_property
 from inspect import iscoroutinefunction
@@ -25,15 +27,15 @@
 if t.TYPE_CHECKING:
     import typing_extensions as te
 
-    F = t.TypeVar("F", bound=t.Callable[..., t.Any])
+    F = t.TypeVar("F", bound=c.Callable[..., t.Any])
     T = t.TypeVar("T")
     P = te.ParamSpec("P")
 
     class PAsyncWrapper(t.Protocol):
-        def __call__(self, f: t.Callable[P, t.Awaitable[T]]) -> t.Callable[P, T]: ...
+        def __call__(self, f: c.Callable[P, c.Awaitable[T]]) -> c.Callable[P, T]: ...
 
     class PSyncWrapper(t.Protocol):
-        def __call__(self, f: t.Callable[P, T]) -> t.Callable[P, t.Awaitable[T]]: ...
+        def __call__(self, f: c.Callable[P, T]) -> c.Callable[P, c.Awaitable[T]]: ...
 
 
 ANY = Symbol("ANY")
@@ -98,7 +100,7 @@ def __init__(self, doc: str | None = None) -> None:
         #: of the mapping is useful as an extremely efficient check to see if
         #: any receivers are connected to the signal.
         self.receivers: dict[
-            t.Any, weakref.ref[t.Callable[..., t.Any]] | t.Callable[..., t.Any]
+            t.Any, weakref.ref[c.Callable[..., t.Any]] | c.Callable[..., t.Any]
         ] = {}
         self.is_muted: bool = False
         self._by_receiver: dict[t.Any, set[t.Any]] = defaultdict(self.set_class)
@@ -166,7 +168,7 @@ def connect(self, receiver: F, sender: t.Any = ANY, weak: bool = True) -> F:
 
         return receiver
 
-    def connect_via(self, sender: t.Any, weak: bool = False) -> t.Callable[[F], F]:
+    def connect_via(self, sender: t.Any, weak: bool = False) -> c.Callable[[F], F]:
         """Connect the decorated function as a receiver for *sender*.
 
         :param sender: Any object or :obj:`ANY`.  The decorated function
@@ -195,8 +197,8 @@ def decorator(fn: F) -> F:
 
     @contextmanager
     def connected_to(
-        self, receiver: t.Callable[..., t.Any], sender: t.Any = ANY
-    ) -> t.Generator[None, None, None]:
+        self, receiver: c.Callable[..., t.Any], sender: t.Any = ANY
+    ) -> c.Generator[None, None, None]:
         """Execute a block with the signal temporarily connected to *receiver*.
 
         :param receiver: a receiver callable
@@ -224,7 +226,7 @@ def connected_to(
             self.disconnect(receiver)
 
     @contextmanager
-    def muted(self) -> t.Generator[None, None, None]:
+    def muted(self) -> c.Generator[None, None, None]:
         """Context manager for temporarily disabling signal.
         Useful for test purposes.
         """
@@ -236,8 +238,8 @@ def muted(self) -> t.Generator[None, None, None]:
             self.is_muted = False
 
     def temporarily_connected_to(
-        self, receiver: t.Callable[..., t.Any], sender: t.Any = ANY
-    ) -> t.ContextManager[None]:
+        self, receiver: c.Callable[..., t.Any], sender: t.Any = ANY
+    ) -> AbstractContextManager[None]:
         """An alias for :meth:`connected_to`.
 
         :param receiver: a receiver callable
@@ -263,7 +265,7 @@ def send(
         *,
         _async_wrapper: PAsyncWrapper | None = None,
         **kwargs: t.Any,
-    ) -> list[tuple[t.Callable[..., t.Any], t.Any]]:
+    ) -> list[tuple[c.Callable[..., t.Any], t.Any]]:
         """Emit this signal on behalf of *sender*, passing on ``kwargs``.
 
         Returns a list of 2-tuples, pairing receivers with their return
@@ -301,7 +303,7 @@ async def send_async(
         *,
         _sync_wrapper: PSyncWrapper | None = None,
         **kwargs: t.Any,
-    ) -> list[tuple[t.Callable[..., t.Any], t.Any]]:
+    ) -> list[tuple[c.Callable[..., t.Any], t.Any]]:
         """Emit this signal on behalf of *sender*, passing on ``kwargs``.
 
         Returns a list of 2-tuples, pairing receivers with their return
@@ -353,7 +355,7 @@ def has_receivers_for(self, sender: t.Any) -> bool:
 
     def receivers_for(
         self, sender: t.Any
-    ) -> t.Generator[t.Callable[..., t.Any], None, None]:
+    ) -> c.Generator[c.Callable[..., t.Any], None, None]:
         """Iterate all live receivers listening for *sender*."""
         # TODO: test receivers_for(ANY)
         if not self.receivers:
@@ -383,7 +385,7 @@ def receivers_for(
             else:
                 yield receiver
 
-    def disconnect(self, receiver: t.Callable[..., t.Any], sender: t.Any = ANY) -> None:
+    def disconnect(self, receiver: c.Callable[..., t.Any], sender: t.Any = ANY) -> None:
         """Disconnect *receiver* from this signal's events.
 
         :param receiver: a previously :meth:`connected<connect>` callable
@@ -392,7 +394,7 @@ def disconnect(self, receiver: t.Callable[..., t.Any], sender: t.Any = ANY) -> N
           to disconnect from all senders.  Defaults to ``ANY``.
 
         """
-        sender_id: t.Hashable
+        sender_id: c.Hashable
 
         if sender is ANY:
             sender_id = ANY_ID
@@ -408,7 +410,7 @@ def disconnect(self, receiver: t.Callable[..., t.Any], sender: t.Any = ANY) -> N
         ):
             self.receiver_disconnected.send(self, receiver=receiver, sender=sender)
 
-    def _disconnect(self, receiver_id: t.Hashable, sender_id: t.Hashable) -> None:
+    def _disconnect(self, receiver_id: c.Hashable, sender_id: c.Hashable) -> None:
         if sender_id == ANY_ID:
             if self._by_receiver.pop(receiver_id, None) is not None:
                 for bucket in self._by_sender.values():
@@ -420,18 +422,18 @@ def _disconnect(self, receiver_id: t.Hashable, sender_id: t.Hashable) -> None:
             self._by_receiver[receiver_id].discard(sender_id)
 
     def _make_cleanup_receiver(
-        self, receiver_id: t.Hashable
-    ) -> t.Callable[[weakref.ref[t.Callable[..., t.Any]]], None]:
+        self, receiver_id: c.Hashable
+    ) -> c.Callable[[weakref.ref[c.Callable[..., t.Any]]], None]:
         """Disconnect a receiver from all senders."""
 
-        def cleanup(ref: weakref.ref[t.Callable[..., t.Any]]) -> None:
+        def cleanup(ref: weakref.ref[c.Callable[..., t.Any]]) -> None:
             self._disconnect(receiver_id, ANY_ID)
 
         return cleanup
 
     def _make_cleanup_sender(
-        self, sender_id: t.Hashable
-    ) -> t.Callable[[weakref.ref[t.Any]], None]:
+        self, sender_id: c.Hashable
+    ) -> c.Callable[[weakref.ref[t.Any]], None]:
         """Disconnect all receivers from a sender."""
         assert sender_id != ANY_ID
 

tests/test_signals.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import collections.abc as c
 import gc
 import sys
 import typing as t
@@ -18,7 +19,7 @@ def collect_acyclic_refs() -> None:
 class Sentinel(list):  # type: ignore[type-arg]
     """A signal receipt accumulator."""
 
-    def make_receiver(self, key: t.Any) -> t.Callable[..., t.Any]:
+    def make_receiver(self, key: t.Any) -> c.Callable[..., t.Any]:
         """Return a generic signal receiver function logging as *key*
 
         When connected to a signal, appends (key, sender, kw) to the Sentinel.
@@ -265,7 +266,7 @@ async def received_async(sender: t.Any) -> None:
     def received(sender: t.Any) -> None:
         sentinel.append(sender)
 
-    def wrapper(func: t.Callable[..., t.Any]) -> t.Callable[..., None]:
+    def wrapper(func: c.Callable[..., t.Any]) -> c.Callable[..., None]:
         async def inner(*args: t.Any, **kwargs: t.Any) -> None:
             func(*args, **kwargs)