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 each update()
• 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

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/except and 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 EventBus into services.
• Untyped payload dicts - typos in keys fail silently. Fix: use dataclass events or TypedDict.

Alternatives

FAQs

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

External Related Links

• Real Python - In-depth tutorials, courses, and practical guides.
• The Hitchhiker's Guide to Python - Best-practices handbook for installation, configuration, style, and tooling.
• Discuss Python (Official Discourse Forum) - Official community forum.
• Fluent Python (2nd Edition) - Intermediate to advanced Python idioms (Luciano Ramalho).
• Python Crash Course (3rd Edition) - Beginner-friendly projects (Eric Matthes).

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+.