Signal class (third-party)¶

def connect(self, receiver: F, sender: t.Any = ANY, weak: bool = True) -> F:
    """Connect ``receiver`` to be called when the signal is sent by
    ``sender``.

    :param receiver: The callable to call when :meth:`send` is called with
        the given ``sender``, passing ``sender`` as a positional argument
        along with any extra keyword arguments.
    :param sender: Any object or :data:`ANY`. ``receiver`` will only be
        called when :meth:`send` is called with this sender. If ``ANY``, the
        receiver will be called for any sender. A receiver may be connected
        to multiple senders by calling :meth:`connect` multiple times.
    :param weak: Track the receiver with a :mod:`weakref`. The receiver will
        be automatically disconnected when it is garbage collected. When
        connecting a receiver defined within a function, set to ``False``,
        otherwise it will be disconnected when the function scope ends.
    """
    receiver_id = make_id(receiver)
    sender_id = ANY_ID if sender is ANY else make_id(sender)

    if weak:
        self.receivers[receiver_id] = make_ref(
            receiver, self._make_cleanup_receiver(receiver_id)
        )
    else:
        self.receivers[receiver_id] = receiver

    self._by_sender[sender_id].add(receiver_id)
    self._by_receiver[receiver_id].add(sender_id)

    if sender is not ANY and sender_id not in self._weak_senders:
        # store a cleanup for weakref-able senders
        try:
            self._weak_senders[sender_id] = make_ref(
                sender, self._make_cleanup_sender(sender_id)
            )
        except TypeError:
            pass

    if "receiver_connected" in self.__dict__ and self.receiver_connected.receivers:
        try:
            self.receiver_connected.send(
                self, receiver=receiver, sender=sender, weak=weak
            )
        except TypeError:
            # TODO no explanation or test for this
            self.disconnect(receiver, sender)
            raise

    return receiver

def connect_via(self, sender: t.Any, weak: bool = False) -> c.Callable[[F], F]:
    """Connect the decorated function to be called when the signal is sent
    by ``sender``.

    The decorated function will be called when :meth:`send` is called with
    the given ``sender``, passing ``sender`` as a positional argument along
    with any extra keyword arguments.

    :param sender: Any object or :data:`ANY`. ``receiver`` will only be
        called when :meth:`send` is called with this sender. If ``ANY``, the
        receiver will be called for any sender. A receiver may be connected
        to multiple senders by calling :meth:`connect` multiple times.
    :param weak: Track the receiver with a :mod:`weakref`. The receiver will
        be automatically disconnected when it is garbage collected. When
        connecting a receiver defined within a function, set to ``False``,
        otherwise it will be disconnected when the function scope ends.=

    .. versionadded:: 1.1
    """

    def decorator(fn: F) -> F:
        self.connect(fn, sender, weak)
        return fn

    return decorator

@contextmanager
def connected_to(
    self, receiver: c.Callable[..., t.Any], sender: t.Any = ANY
) -> c.Generator[None, None, None]:
    """A context manager that temporarily connects ``receiver`` to the
    signal while a ``with`` block executes. When the block exits, the
    receiver is disconnected. Useful for tests.

    :param receiver: The callable to call when :meth:`send` is called with
        the given ``sender``, passing ``sender`` as a positional argument
        along with any extra keyword arguments.
    :param sender: Any object or :data:`ANY`. ``receiver`` will only be
        called when :meth:`send` is called with this sender. If ``ANY``, the
        receiver will be called for any sender.

    .. versionadded:: 1.1
    """
    self.connect(receiver, sender=sender, weak=False)

    try:
        yield None
    finally:
        self.disconnect(receiver)

@contextmanager
def muted(self) -> c.Generator[None, None, None]:
    """A context manager that temporarily disables the signal. No receivers
    will be called if the signal is sent, until the ``with`` block exits.
    Useful for tests.
    """
    self.is_muted = True

    try:
        yield None
    finally:
        self.is_muted = False

def send(
    self,
    sender: t.Any | None = None,
    /,
    *,
    _async_wrapper: c.Callable[
        [c.Callable[..., c.Coroutine[t.Any, t.Any, t.Any]]], c.Callable[..., t.Any]
    ]
    | None = None,
    **kwargs: t.Any,
) -> list[tuple[c.Callable[..., t.Any], t.Any]]:
    """Call all receivers that are connected to the given ``sender``
    or :data:`ANY`. Each receiver is called with ``sender`` as a positional
    argument along with any extra keyword arguments. Return a list of
    ``(receiver, return value)`` tuples.

    The order receivers are called is undefined, but can be influenced by
    setting :attr:`set_class`.

    If a receiver raises an exception, that exception will propagate up.
    This makes debugging straightforward, with an assumption that correctly
    implemented receivers will not raise.

    :param sender: Call receivers connected to this sender, in addition to
        those connected to :data:`ANY`.
    :param _async_wrapper: Will be called on any receivers that are async
        coroutines to turn them into sync callables. For example, could run
        the receiver with an event loop.
    :param kwargs: Extra keyword arguments to pass to each receiver.

    .. versionchanged:: 1.7
        Added the ``_async_wrapper`` argument.
    """
    if self.is_muted:
        return []

    results = []

    for receiver in self.receivers_for(sender):
        if iscoroutinefunction(receiver):
            if _async_wrapper is None:
                raise RuntimeError("Cannot send to a coroutine function.")

            result = _async_wrapper(receiver)(sender, **kwargs)
        else:
            result = receiver(sender, **kwargs)

        results.append((receiver, result))

    return results

async def send_async(
    self,
    sender: t.Any | None = None,
    /,
    *,
    _sync_wrapper: c.Callable[
        [c.Callable[..., t.Any]], c.Callable[..., c.Coroutine[t.Any, t.Any, t.Any]]
    ]
    | None = None,
    **kwargs: t.Any,
) -> list[tuple[c.Callable[..., t.Any], t.Any]]:
    """Await all receivers that are connected to the given ``sender``
    or :data:`ANY`. Each receiver is called with ``sender`` as a positional
    argument along with any extra keyword arguments. Return a list of
    ``(receiver, return value)`` tuples.

    The order receivers are called is undefined, but can be influenced by
    setting :attr:`set_class`.

    If a receiver raises an exception, that exception will propagate up.
    This makes debugging straightforward, with an assumption that correctly
    implemented receivers will not raise.

    :param sender: Call receivers connected to this sender, in addition to
        those connected to :data:`ANY`.
    :param _sync_wrapper: Will be called on any receivers that are sync
        callables to turn them into async coroutines. For example,
        could call the receiver in a thread.
    :param kwargs: Extra keyword arguments to pass to each receiver.

    .. versionadded:: 1.7
    """
    if self.is_muted:
        return []

    results = []

    for receiver in self.receivers_for(sender):
        if not iscoroutinefunction(receiver):
            if _sync_wrapper is None:
                raise RuntimeError("Cannot send to a non-coroutine function.")

            result = await _sync_wrapper(receiver)(sender, **kwargs)
        else:
            result = await receiver(sender, **kwargs)

        results.append((receiver, result))

    return results

def has_receivers_for(self, sender: t.Any) -> bool:
    """Check if there is at least one receiver that will be called with the
    given ``sender``. A receiver connected to :data:`ANY` will always be
    called, regardless of sender. Does not check if weakly referenced
    receivers are still live. See :meth:`receivers_for` for a stronger
    search.

    :param sender: Check for receivers connected to this sender, in addition
        to those connected to :data:`ANY`.
    """
    if not self.receivers:
        return False

    if self._by_sender[ANY_ID]:
        return True

    if sender is ANY:
        return False

    return make_id(sender) in self._by_sender

def receivers_for(
    self, sender: t.Any
) -> c.Generator[c.Callable[..., t.Any], None, None]:
    """Yield each receiver to be called for ``sender``, in addition to those
    to be called for :data:`ANY`. Weakly referenced receivers that are not
    live will be disconnected and skipped.

    :param sender: Yield receivers connected to this sender, in addition
        to those connected to :data:`ANY`.
    """
    # TODO: test receivers_for(ANY)
    if not self.receivers:
        return

    sender_id = make_id(sender)

    if sender_id in self._by_sender:
        ids = self._by_sender[ANY_ID] | self._by_sender[sender_id]
    else:
        ids = self._by_sender[ANY_ID].copy()

    for receiver_id in ids:
        receiver = self.receivers.get(receiver_id)

        if receiver is None:
            continue

        if isinstance(receiver, weakref.ref):
            strong = receiver()

            if strong is None:
                self._disconnect(receiver_id, ANY_ID)
                continue

            yield strong
        else:
            yield receiver

def disconnect(self, receiver: c.Callable[..., t.Any], sender: t.Any = ANY) -> None:
    """Disconnect ``receiver`` from being called when the signal is sent by
    ``sender``.

    :param receiver: A connected receiver callable.
    :param sender: Disconnect from only this sender. By default, disconnect
        from all senders.
    """
    sender_id: c.Hashable

    if sender is ANY:
        sender_id = ANY_ID
    else:
        sender_id = make_id(sender)

    receiver_id = make_id(receiver)
    self._disconnect(receiver_id, sender_id)

    if (
        "receiver_disconnected" in self.__dict__
        and self.receiver_disconnected.receivers
    ):
        self.receiver_disconnected.send(self, receiver=receiver, sender=sender)