CtrlK
BlogDocsLog inGet started
Tessl Logo

tessl/pypi-aiosignal

A coroutine-based signal implementation for asyncio projects providing asynchronous callback management

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview
Eval results
Files

index.mddocs/

aiosignal

A coroutine-based signal implementation for asyncio projects. aiosignal provides a Signal class that acts as a list of registered asynchronous callbacks with a two-stage lifecycle: first, callbacks can be registered using standard list operations, then the signal is frozen to prevent further modifications and enable callback execution.

Package Information

  • Package Name: aiosignal
  • Package Type: pypi
  • Language: Python
  • Installation: pip install aiosignal
  • Minimum Python Version: 3.9

Core Imports

from aiosignal import Signal

Version information:

from aiosignal import __version__
print(__version__)  # "1.4.0"

Basic Usage

import asyncio
from aiosignal import Signal

# Create a signal with an owner object
class MyApp:
    def __init__(self):
        self.on_startup = Signal(self)

app = MyApp()

# Register callbacks using list operations
async def init_database():
    print("Database initialized")

async def setup_logging():
    print("Logging configured")

app.on_startup.append(init_database)
app.on_startup.append(setup_logging)

# Or register using decorator syntax
@app.on_startup
async def load_config():
    print("Configuration loaded")

# Freeze the signal to enable execution
app.on_startup.freeze()

# Send the signal to execute all callbacks
async def main():
    await app.on_startup.send()

# Run the example
asyncio.run(main())

Architecture

The Signal class inherits from FrozenList (from the frozenlist package), providing a two-stage lifecycle:

  1. Registration Stage: Signal acts as a mutable list where callbacks can be added, removed, or modified using standard list operations
  2. Execution Stage: After calling freeze(), the signal becomes immutable and can execute callbacks via send()

This design ensures thread-safe callback execution while preventing accidental modifications during execution.

Capabilities

Signal Creation and Management

Create and configure signal instances for managing asynchronous callbacks.

class Signal(FrozenList[Callable[[Unpack[_Ts]], Awaitable[object]]]):
    def __init__(self, owner: object):
        """
        Initialize a signal with an owner object.
        
        Args:
            owner: The object that owns this signal (for identification)
        """
    
    def __repr__(self) -> str:
        """
        Return string representation showing owner, frozen state, and callbacks.
        
        Returns:
            str: Formatted representation
        """

Callback Registration

Register asynchronous callbacks that will be executed when the signal is sent.

def __call__(
    self, func: Callable[[Unpack[_Ts]], Awaitable[_T]]
) -> Callable[[Unpack[_Ts]], Awaitable[_T]]:
    """
    Decorator to add a function to this signal.
    
    Args:
        func: Asynchronous function to register as callback
        
    Returns:
        The same function (for use as decorator)
    """

All standard list operations are available for callback management:

  • append(callback) - Add callback to signal
  • insert(index, callback) - Insert callback at specific position
  • remove(callback) - Remove specific callback
  • pop(index) - Remove and return callback at index
  • clear() - Remove all callbacks
  • __getitem__(index) - Get callback at index
  • __setitem__(index, callback) - Replace callback at index
  • __delitem__(index) - Delete callback at index
  • __len__() - Get number of callbacks
  • __iter__() - Iterate over callbacks

Signal Execution

Execute all registered callbacks by sending the signal with optional arguments.

async def send(self, *args: Unpack[_Ts], **kwargs: Any) -> None:
    """
    Send data to all registered receivers.
    
    The signal must be frozen before sending. All registered callbacks
    will be executed in registration order.
    
    Args:
        *args: Positional arguments to pass to each callback
        **kwargs: Keyword arguments to pass to each callback
        
    Raises:
        RuntimeError: If signal is not frozen
        TypeError: If any callback is not a coroutine function
    """

Signal State Management

Control the signal's lifecycle and state transitions.

def freeze(self) -> None:
    """
    Freeze the signal to prevent further modifications.
    
    After freezing, callbacks cannot be added, removed, or modified.
    The signal can then be used to send data to callbacks.
    
    Raises:
        RuntimeError: If attempting to modify after freezing
    """

@property
def frozen(self) -> bool:
    """
    Check if signal is frozen.
    
    Returns:
        bool: True if signal is frozen, False otherwise
    """

Types

from typing import Any, Awaitable, Callable, TypeVar
from typing_extensions import Unpack, TypeVarTuple

_T = TypeVar("_T")
_Ts = TypeVarTuple("_Ts", default=Unpack[tuple[()]])

__version__: str = "1.4.0"

class Signal(FrozenList[Callable[[Unpack[_Ts]], Awaitable[object]]]):
    """
    Generic signal class that can be parameterized with callback argument types.
    
    Examples:
        Signal[str, int] - callbacks take (str, int) arguments
        Signal[dict] - callbacks take dict argument
        Signal - callbacks take any arguments
    """
    
    _owner: object

Error Handling

aiosignal raises the following exceptions:

  • RuntimeError: When attempting to send a non-frozen signal or modify a frozen signal
  • TypeError: When registered callbacks are not coroutine functions
# Example error handling
import asyncio
from aiosignal import Signal

signal = Signal(owner=None)

# This raises RuntimeError - signal not frozen
try:
    await signal.send()
except RuntimeError as e:
    print(f"Cannot send non-frozen signal: {e}")

# Register non-coroutine callback
def sync_callback():
    pass

signal.append(sync_callback)
signal.freeze()

# This raises TypeError - callback not a coroutine
try:
    await signal.send()
except TypeError as e:
    print(f"Callback must be coroutine: {e}")

Dependencies

aiosignal requires the following dependencies:

  • frozenlist >= 1.1.0: Provides the immutable list base class
  • typing-extensions >= 4.2: Advanced typing features for Python < 3.13

The package is designed for Python 3.9+ with full asyncio compatibility.

Install with Tessl CLI

npx tessl i tessl/pypi-aiosignal

docs

index.md

tile.json