0
# aiosignal
1

2
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.
3

4
## Package Information
5

6
- **Package Name**: aiosignal
7
- **Package Type**: pypi
8
- **Language**: Python
9
- **Installation**: `pip install aiosignal`
10
- **Minimum Python Version**: 3.9
11

12
## Core Imports
13

14
```python
15
from aiosignal import Signal
16
```
17

18
Version information:
19

20
```python
21
from aiosignal import __version__
22
print(__version__)  # "1.4.0"
23
```
24

25
## Basic Usage
26

27
```python
28
import asyncio
29
from aiosignal import Signal
30

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

36
app = MyApp()
37

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

42
async def setup_logging():
43
    print("Logging configured")
44

45
app.on_startup.append(init_database)
46
app.on_startup.append(setup_logging)
47

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

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

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

60
# Run the example
61
asyncio.run(main())
62
```
63

64
## Architecture
65

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

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

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

73
## Capabilities
74

75
### Signal Creation and Management
76

77
Create and configure signal instances for managing asynchronous callbacks.
78

79
```python { .api }
80
class Signal(FrozenList[Callable[[Unpack[_Ts]], Awaitable[object]]]):
81
    def __init__(self, owner: object):
82
        """
83
        Initialize a signal with an owner object.
84
        
85
        Args:
86
            owner: The object that owns this signal (for identification)
87
        """
88
    
89
    def __repr__(self) -> str:
90
        """
91
        Return string representation showing owner, frozen state, and callbacks.
92
        
93
        Returns:
94
            str: Formatted representation
95
        """
96
```
97

98
### Callback Registration
99

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

102
```python { .api }
103
def __call__(
104
    self, func: Callable[[Unpack[_Ts]], Awaitable[_T]]
105
) -> Callable[[Unpack[_Ts]], Awaitable[_T]]:
106
    """
107
    Decorator to add a function to this signal.
108
    
109
    Args:
110
        func: Asynchronous function to register as callback
111
        
112
    Returns:
113
        The same function (for use as decorator)
114
    """
115
```
116

117
All standard list operations are available for callback management:
118

119
- `append(callback)` - Add callback to signal
120
- `insert(index, callback)` - Insert callback at specific position  
121
- `remove(callback)` - Remove specific callback
122
- `pop(index)` - Remove and return callback at index
123
- `clear()` - Remove all callbacks
124
- `__getitem__(index)` - Get callback at index
125
- `__setitem__(index, callback)` - Replace callback at index
126
- `__delitem__(index)` - Delete callback at index
127
- `__len__()` - Get number of callbacks
128
- `__iter__()` - Iterate over callbacks
129

130
### Signal Execution
131

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

134
```python { .api }
135
async def send(self, *args: Unpack[_Ts], **kwargs: Any) -> None:
136
    """
137
    Send data to all registered receivers.
138
    
139
    The signal must be frozen before sending. All registered callbacks
140
    will be executed in registration order.
141
    
142
    Args:
143
        *args: Positional arguments to pass to each callback
144
        **kwargs: Keyword arguments to pass to each callback
145
        
146
    Raises:
147
        RuntimeError: If signal is not frozen
148
        TypeError: If any callback is not a coroutine function
149
    """
150
```
151

152
### Signal State Management
153

154
Control the signal's lifecycle and state transitions.
155

156
```python { .api }
157
def freeze(self) -> None:
158
    """
159
    Freeze the signal to prevent further modifications.
160
    
161
    After freezing, callbacks cannot be added, removed, or modified.
162
    The signal can then be used to send data to callbacks.
163
    
164
    Raises:
165
        RuntimeError: If attempting to modify after freezing
166
    """
167

168
@property
169
def frozen(self) -> bool:
170
    """
171
    Check if signal is frozen.
172
    
173
    Returns:
174
        bool: True if signal is frozen, False otherwise
175
    """
176
```
177

178
## Types
179

180
```python { .api }
181
from typing import Any, Awaitable, Callable, TypeVar
182
from typing_extensions import Unpack, TypeVarTuple
183

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

187
__version__: str = "1.4.0"
188

189
class Signal(FrozenList[Callable[[Unpack[_Ts]], Awaitable[object]]]):
190
    """
191
    Generic signal class that can be parameterized with callback argument types.
192
    
193
    Examples:
194
        Signal[str, int] - callbacks take (str, int) arguments
195
        Signal[dict] - callbacks take dict argument
196
        Signal - callbacks take any arguments
197
    """
198
    
199
    _owner: object
200
```
201

202
## Error Handling
203

204
aiosignal raises the following exceptions:
205

206
- **RuntimeError**: When attempting to send a non-frozen signal or modify a frozen signal
207
- **TypeError**: When registered callbacks are not coroutine functions
208

209
```python
210
# Example error handling
211
import asyncio
212
from aiosignal import Signal
213

214
signal = Signal(owner=None)
215

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

222
# Register non-coroutine callback
223
def sync_callback():
224
    pass
225

226
signal.append(sync_callback)
227
signal.freeze()
228

229
# This raises TypeError - callback not a coroutine
230
try:
231
    await signal.send()
232
except TypeError as e:
233
    print(f"Callback must be coroutine: {e}")
234
```
235

236
## Dependencies
237

238
aiosignal requires the following dependencies:
239

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

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