Very basic event publishing system with synchronous dispatching for Python applications
npx @tessl/cli install tessl/pypi-zope-event@5.1.00
# zope.event
1
2
A minimalist event system providing simple synchronous event publishing for Python applications. The package enables publish-subscribe patterns where applications can notify subscribers of events without requiring knowledge of who is listening, serving as a foundation for building more sophisticated event dispatching systems.
3
4
## Package Information
5
6
- **Package Name**: zope.event
7
- **Language**: Python
8
- **Installation**: `pip install zope.event`
9
- **Python Versions**: 3.9+
10
11
## Core Imports
12
13
```python
14
import zope.event
15
```
16
17
For direct access to core functions:
18
19
```python
20
from zope.event import notify, subscribers
21
```
22
23
For class-based event handling:
24
25
```python
26
from zope.event.classhandler import handler
27
```
28
29
## Basic Usage
30
31
```python
32
import zope.event
33
34
# Define a simple event handler
35
def my_handler(event):
36
print(f"Received event: {event}")
37
38
# Register the handler
39
zope.event.subscribers.append(my_handler)
40
41
# Create and publish an event
42
event_data = {"type": "user_login", "user_id": 123}
43
zope.event.notify(event_data)
44
# Output: Received event: {'type': 'user_login', 'user_id': 123}
45
46
# Class-based event handling
47
from zope.event.classhandler import handler
48
49
class UserLoginEvent:
50
def __init__(self, user_id):
51
self.user_id = user_id
52
53
# Register handler for specific event class
54
@handler(UserLoginEvent)
55
def handle_login(event):
56
print(f"User {event.user_id} logged in")
57
58
# Publish class-based event
59
login_event = UserLoginEvent(456)
60
zope.event.notify(login_event)
61
# Output: User 456 logged in
62
```
63
64
## Capabilities
65
66
### Event Publishing
67
68
Core event publishing functionality that allows applications to notify all registered subscribers of events synchronously.
69
70
```python { .api }
71
def notify(event):
72
"""
73
Notify all subscribers of the given event.
74
75
Args:
76
event: Any object to be published as an event
77
78
Returns:
79
None
80
81
Raises:
82
Any exception raised by subscribers is propagated without
83
running remaining subscribers
84
"""
85
```
86
87
### Subscriber Management
88
89
Global registry for event subscribers that any callable can be added to for receiving event notifications.
90
91
```python { .api }
92
subscribers: list
93
"""
94
Global list of event subscribers.
95
96
Applications append callable subscribers to this list. Each subscriber
97
must accept a single argument (the event object). Subscribers are called
98
in the order they appear in the list.
99
100
Example:
101
def my_handler(event):
102
print(f"Got event: {event}")
103
104
zope.event.subscribers.append(my_handler)
105
"""
106
```
107
108
### Class-Based Event Handling
109
110
Enhanced event handling system that registers handlers for specific event classes, supporting inheritance and method resolution order dispatching.
111
112
```python { .api }
113
def handler(event_class, handler_=None, _decorator=False):
114
"""
115
Register an event handler for a specific event class.
116
117
Can be used as a decorator or called directly with class and handler.
118
Handlers are called based on the event object's class hierarchy using
119
method resolution order.
120
121
Args:
122
event_class: type - the event class to handle
123
handler_: callable, optional - the handler function
124
_decorator: bool, internal - indicates decorator usage
125
126
Returns:
127
callable: decorator function when handler_ is None
128
callable: the handler function when _decorator is True
129
None: when called directly with both event_class and handler_
130
131
Examples:
132
# Direct registration
133
handler(MyEvent, my_handler)
134
135
# Decorator usage
136
@handler(MyEvent)
137
def my_handler(event):
138
pass
139
"""
140
141
def dispatch(event):
142
"""
143
Internal dispatcher function that calls handlers for event classes.
144
145
This function is automatically registered as a subscriber when the first
146
handler is registered. It iterates through the event's class hierarchy
147
(method resolution order) and calls all registered handlers for each class.
148
149
Args:
150
event: Any - the event object to dispatch
151
152
Returns:
153
None
154
"""
155
156
# Global registry of event class handlers
157
registry: dict
158
"""
159
Internal registry mapping event classes to their handler lists.
160
161
This dictionary maps event classes (types) to lists of handler functions.
162
Handlers are stored in registration order and called in that order for
163
each class in the event's method resolution order.
164
165
Type: Dict[type, List[Callable[[Any], None]]]
166
"""
167
```
168
169
## Types
170
171
```python { .api }
172
from typing import Any, Callable, Dict, List, Optional, Union
173
174
# Event objects can be any Python object
175
Event = Any
176
177
# Subscriber functions must match this signature
178
Subscriber = Callable[[Any], None]
179
180
# Event class for class-based handling
181
EventClass = type
182
183
# Handler function signature
184
Handler = Callable[[Any], None]
185
186
# Registry type for internal use
187
Registry = Dict[type, List[Handler]]
188
189
# Handler decorator return type
190
HandlerDecorator = Callable[[Handler], Handler]
191
```
192
193
## Error Handling
194
195
- **Subscriber exceptions**: Any exception raised by a subscriber during `notify()` is propagated immediately without calling remaining subscribers
196
- **Handler registration**: The class-based handler system automatically registers its dispatcher with the global subscribers list on first use
197
- **Event inheritance**: Class-based handlers are called in method resolution order, supporting proper inheritance patterns
198
199
## Advanced Usage
200
201
### Multiple Event Types
202
203
```python
204
import zope.event
205
from zope.event.classhandler import handler
206
207
# Define event hierarchy
208
class BaseEvent:
209
pass
210
211
class UserEvent(BaseEvent):
212
def __init__(self, user_id):
213
self.user_id = user_id
214
215
class LoginEvent(UserEvent):
216
pass
217
218
class LogoutEvent(UserEvent):
219
pass
220
221
# Register handlers at different levels
222
@handler(BaseEvent)
223
def log_all_events(event):
224
print(f"Event: {event.__class__.__name__}")
225
226
@handler(UserEvent)
227
def track_user_activity(event):
228
print(f"User activity: {event.user_id}")
229
230
@handler(LoginEvent)
231
def handle_login(event):
232
print(f"Login: {event.user_id}")
233
234
# Publishing events triggers appropriate handlers
235
zope.event.notify(LoginEvent(123))
236
# Output (handlers called in MRO order, then registration order):
237
# Login: 123
238
# User activity: 123
239
# Event: LoginEvent
240
```
241
242
### Custom Subscriber Management
243
244
```python
245
import zope.event
246
247
# Save and restore subscriber state
248
original_subscribers = zope.event.subscribers[:]
249
zope.event.subscribers.clear()
250
251
# Add temporary subscribers
252
def temp_handler(event):
253
print("Temporary handler")
254
255
zope.event.subscribers.append(temp_handler)
256
257
# Use the system
258
zope.event.notify("test event")
259
260
# Restore original state
261
zope.event.subscribers[:] = original_subscribers
262
```