Observer & Pub/Sub
Observer notifies interested parties when state changes. Pub/Sub routes messages through topics so publishers do not know subscribers. Python implements both with callables, list/set of handlers, or libraries like blinker - no heavyweight framework required.
Recipe
Quick-reference recipe card - copy-paste ready.
from collections import defaultdict
from typing import Callable
EventHandler = Callable[[str, dict], None]
class EventBus:
def __init__(self) -> None:
self._subs: dict[str, list[EventHandler]] = defaultdict(list)
def subscribe(self, topic: str, handler: EventHandler) -> None:
self._subs[topic].append(handler)
def publish(self, topic: str, payload: dict) -> None:
for handler in list(self._subs[topic]):
handler(topic, payload)When to reach for this:
- Domain events - order placed, user registered, cache invalidated
- UI decoupling - view listens, model publishes (desktop, game loops)
- Cross-module hooks - plugins subscribe without importing publishers
- Async fan-out - multiple handlers react to one signal
- Testing - assert handlers received expected events
Working Example
An order service publishes events; audit and email handlers subscribe independently.
from collections import defaultdict
from dataclasses import dataclass, field
from typing import Callable
Handler = Callable[[str, dict], None]
class EventBus:
def __init__(self) -> None:
self._handlers: dict[str, list[Handler]] = defaultdict(list)
def on(self, event: str, handler: Handler) -> None:
self._handlers[event].append(handler)
def emit(self, event: str, **payload) -> None:
for handler in list(self._handlers[event]):
handler(event, payload)
@dataclass
class OrderService:
bus: EventBus
orders: list[dict] = field(default_factory=list)
def place(self, sku: str, qty: int) -> dict:
order = {"sku": sku, "qty": qty}
self.orders.append(order)
self.bus.emit("order.placed", order=order)
return order
def audit_log(_event: str, payload: dict) -> None:
print("AUDIT", payload)
def send_email(_event: str, payload: dict) -> None:
print("EMAIL confirmation for", payload["order"]["sku"])
bus = EventBus()
bus.on("order.placed", audit_log)
bus.on("order.placed", send_email)
svc = OrderService(bus)
svc.place("BOOK-1", 2)What this demonstrates:
- Publishers emit named events with keyword payload
- Multiple subscribers per topic without tight coupling
list()copy during emit avoids mutation-during-iteration bugs- Handlers are plain functions - easy to test in isolation
Deep Dive
How It Works
- Observer: subject maintains a list of observers;
notify()calls eachupdate() - Pub/Sub: bus maps topic strings to handler lists; publishers do not hold subscriber references
- Handlers should be fast; heavy work goes to queues or background tasks
- Unsubscribe removes handlers to prevent leaks on long-lived buses
- Typed events can use dataclasses instead of loose dicts
Observer vs Pub/Sub
| Style | Coupling | Typical API |
|---|---|---|
| Observer | Subject knows observer interface | .attach() / .notify() |
| Pub/Sub | Topic string mediates | .subscribe(topic) / .publish() |
Python Notes
# Weak references for auto-cleanup when observer is GC'd
import weakref
class Subject:
def __init__(self) -> None:
self._observers: list = []
def attach(self, obs):
self._observers.append(weakref.ref(obs))Gotchas
- Handler exceptions abort fan-out - one failing subscriber blocks others. Fix: wrap each call in
try/exceptand log. - Memory leaks - forgotten subscriptions keep closures alive. Fix: unsubscribe on teardown or use weakrefs.
- Re-entrant emit - handler publishes same topic while handling. Fix: document reentrancy or queue events.
- Global bus singleton - hidden dependencies everywhere. Fix: inject
EventBusinto services. - Untyped payload dicts - typos in keys fail silently. Fix: use dataclass events or TypedDict.
Alternatives
| Alternative | Use When | Don't Use When |
|---|---|---|
| Direct method calls | Two known collaborators | Many unknown subscribers |
asyncio queues | Async workers consume events | Simple sync callbacks suffice |
| Message broker (Redis, Kafka) | Cross-process distribution | In-process only |
| Signals (Django, blinker) | Framework integration | Minimal deps preferred |
FAQs
What is the difference between Observer and Pub/Sub?
Observer couples to a subject object. Pub/Sub routes by topic through a bus - publishers do not reference subscribers directly.
Should handlers run synchronously?
For in-process domain events, usually yes - keep them fast. Offload I/O to threads, tasks, or queues.
How do I unsubscribe?
Return a disposable token from subscribe, or pass the same function reference to unsubscribe.
Are async handlers supported?
Use an async bus that awaits coroutine handlers, or schedule tasks with asyncio.create_task.
How do I order handler execution?
Document FIFO order from subscription list. Priority queues if order matters for business rules.
What about once handlers?
Wrap with a flag or use functools.lru_cache pattern - remove self after first call.
Can I use typing for events?
Yes. @dataclass(frozen=True) class OrderPlaced: order_id: str plus emit(OrderPlaced(...)) improves safety.
How do I test event flows?
Collect events in a test handler list. Assert topic names and payloads without mocking the entire bus.
Is this the same as Django signals?
Same idea. Django signals are framework-integrated pub/sub; this pattern is stdlib-only.
When does pub/sub become overkill?
Two classes with a clear call graph - a direct method is simpler and easier to trace.
Related
- The Strategy & Dispatch Patterns - route by event type
- Dependency Injection in Python - inject the bus
- Factory & Builder Patterns - construct services with buses
- Pythonic Patterns Basics - section overview
Stack versions: This page was written for Python 3.14.0 (stable 3.14, maintenance 3.13), FastAPI 0.115+, Django 5.2, Flask 3.1, Pydantic 2, PyTorch 2.6+, pandas 2.2+, Polars 1.x, ruff 0.9+, and uv 0.6+.