codekingpro/portable-devtools

1"""2This module provides signals, which are a simple dispatching system that allows any number of interested parties3to subscribe to events ("signals").4 5This is similar to the Blinker library (https://pypi.org/project/blinker/), with the following changes:6  - provides only a small subset of Blinker's functionality7  - supports type hints8  - supports async receivers.9"""10 11from __future__ import annotations12 13import asyncio14import inspect15import weakref16from collections.abc import Awaitable17from collections.abc import Callable18from typing import Any19from typing import cast20from typing import Generic21from typing import ParamSpec22from typing import TypeVar23 24P = ParamSpec("P")25R = TypeVar("R")26 27 28def make_weak_ref(obj: Any) -> weakref.ReferenceType:29    """30    Like weakref.ref(), but using weakref.WeakMethod for bound methods.31    """32    if hasattr(obj, "__self__"):33        return cast(weakref.ref, weakref.WeakMethod(obj))34    else:35        return weakref.ref(obj)36 37 38# We're running into https://github.com/python/mypy/issues/6073 here,39# which is why the base class is a mixin and not a generic superclass.40class _SignalMixin:41    def __init__(self) -> None:42        self.receivers: list[weakref.ref[Callable]] = []43 44    def connect(self, receiver: Callable) -> None:45        """46        Register a signal receiver.47 48        The signal will only hold a weak reference to the receiver function.49        """50        receiver = make_weak_ref(receiver)51        self.receivers.append(receiver)52 53    def disconnect(self, receiver: Callable) -> None:54        self.receivers = [r for r in self.receivers if r() != receiver]55 56    def notify(self, *args, **kwargs):57        cleanup = False58        for ref in self.receivers:59            r = ref()60            if r is not None:61                yield r(*args, **kwargs)62            else:63                cleanup = True64        if cleanup:65            self.receivers = [r for r in self.receivers if r() is not None]66 67 68class _SyncSignal(Generic[P], _SignalMixin):69    def connect(self, receiver: Callable[P, None]) -> None:70        assert not inspect.iscoroutinefunction(receiver)71        super().connect(receiver)72 73    def disconnect(self, receiver: Callable[P, None]) -> None:74        super().disconnect(receiver)75 76    def send(self, *args: P.args, **kwargs: P.kwargs) -> None:77        for ret in super().notify(*args, **kwargs):78            assert ret is None or not inspect.isawaitable(ret)79 80 81class _AsyncSignal(Generic[P], _SignalMixin):82    def connect(self, receiver: Callable[P, Awaitable[None] | None]) -> None:83        super().connect(receiver)84 85    def disconnect(self, receiver: Callable[P, Awaitable[None] | None]) -> None:86        super().disconnect(receiver)87 88    async def send(self, *args: P.args, **kwargs: P.kwargs) -> None:89        await asyncio.gather(90            *[91                aws92                for aws in super().notify(*args, **kwargs)93                if aws is not None and inspect.isawaitable(aws)94            ]95        )96 97 98# noinspection PyPep8Naming99def SyncSignal(receiver_spec: Callable[P, None]) -> _SyncSignal[P]:100    """101    Create a synchronous signal with the given function signature for receivers.102 103    Example:104 105        s = SyncSignal(lambda event: None)  # all receivers must accept a single "event" argument.106        def receiver(event):107            print(event)108 109        s.connect(receiver)110        s.send("foo")  # prints foo111        s.send(event="bar")  # prints bar112 113        def receiver2():114            ...115 116        s.connect(receiver2)  # mypy complains about receiver2 not having the right signature117 118        s2 = SyncSignal(lambda: None)  # this signal has no arguments119        s2.send()120    """121    return cast(_SyncSignal[P], _SyncSignal())122 123 124# noinspection PyPep8Naming125def AsyncSignal(receiver_spec: Callable[P, Awaitable[None] | None]) -> _AsyncSignal[P]:126    """127    Create an signal that supports both regular and async receivers:128 129    Example:130 131        s = AsyncSignal(lambda event: None)132        async def receiver(event):133            print(event)134        s.connect(receiver)135        await s.send("foo")  # prints foo136    """137    return cast(_AsyncSignal[P], _AsyncSignal())138