From facf2c3814859d23a8b264e5348b6f105cde0b3a Mon Sep 17 00:00:00 2001 From: David Lord Date: Sun, 28 Apr 2024 10:13:44 -0700 Subject: [PATCH] use collections.abc instead of typing --- CHANGES.rst | 3 ++- src/blinker/_utilities.py | 7 ++++--- src/blinker/base.py | 44 ++++++++++++++++++++------------------- tests/test_signals.py | 5 +++-- 4 files changed, 32 insertions(+), 27 deletions(-) diff --git a/CHANGES.rst b/CHANGES.rst index d918402..3d260d0 100644 --- a/CHANGES.rst +++ b/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 diff --git a/src/blinker/_utilities.py b/src/blinker/_utilities.py index c948b09..000c902 100644 --- a/src/blinker/_utilities.py +++ b/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] diff --git a/src/blinker/base.py b/src/blinker/base.py index abae4bb..5d001af 100644 --- a/src/blinker/base.py +++ b/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` 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 diff --git a/tests/test_signals.py b/tests/test_signals.py index 5edb536..93a9c48 100644 --- a/tests/test_signals.py +++ b/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)