Skip to content

Instantly share code, notes, and snippets.

@abersheeran
Last active January 16, 2025 06:22
Show Gist options
  • Save abersheeran/7a23bfd2568e5eb5b333eedf9171b81e to your computer and use it in GitHub Desktop.
Save abersheeran/7a23bfd2568e5eb5b333eedf9171b81e to your computer and use it in GitHub Desktop.
Typed asynchronous signal for Python. A few lines of code will do the trick, no need to install another aiosignals.
import asyncio
from typing import Callable, Coroutine, MutableSequence, ParamSpec
P = ParamSpec("P")
R = Coroutine[None, None, None]
class AsyncSignal(MutableSequence[Callable[P, R]]):
def __init__(self, *iterable: Callable[P, R]) -> None:
self.callbacks = list(iterable)
def __len__(self) -> int:
return len(self.callbacks)
def __getitem__(self, index: int) -> Callable[P, R]:
return self.callbacks[index]
def __setitem__(self, index: int, value: Callable[P, R]) -> None:
self.callbacks[index] = value
def __delitem__(self, index: int) -> None:
del self.callbacks[index]
def insert(self, index: int, value: Callable[P, R]) -> None:
self.callbacks.insert(index, value)
def connect(self, callback: Callable[P, R]) -> None:
self.callbacks.append(callback)
def disconnect(self, callback: Callable[P, R]) -> None:
self.callbacks.remove(callback)
async def send(self, *args: P.args, **kwargs: P.kwargs) -> None:
async with asyncio.TaskGroup() as tg:
for callback in self.callbacks:
task = tg.create_task(callback(*args, **kwargs))
task.add_done_callback(lambda _: task.result()) # raise exception
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment