Python Event Bus Patterns — Core Concepts

What an Event Bus Does

An event bus decouples the parts of a program that produce events from the parts that react to them. The producer publishes an event (a data object describing what happened), and the bus routes it to all registered handlers. Neither side knows about the other directly.

This is the Observer pattern scaled up. Instead of each observable managing its own subscriber list, a centralized bus manages all subscriptions.

In-Process Event Bus

The simplest event bus lives inside a single Python process:

from collections import defaultdict
from typing import Callable, Any

class EventBus:
    def __init__(self):
        self._handlers: dict[type, list[Callable]] = defaultdict(list)

    def subscribe(self, event_type: type, handler: Callable):
        self._handlers[event_type].append(handler)

    def publish(self, event: Any):
        for handler in self._handlers[type(event)]:
            handler(event)

Events are regular Python objects — dataclasses work well:

from dataclasses import dataclass

@dataclass
class OrderPlaced:
    order_id: str
    user_id: str
    amount: float

bus = EventBus()
bus.subscribe(OrderPlaced, send_confirmation_email)
bus.subscribe(OrderPlaced, update_inventory)
bus.subscribe(OrderPlaced, log_analytics_event)

bus.publish(OrderPlaced(order_id="123", user_id="u1", amount=49.99))

All three handlers execute when the event is published. The order processing code only knows about EventBus and OrderPlaced — it has no idea that emails, inventory, and analytics exist.

Handler Registration Patterns

Decorator-Based

@bus.on(OrderPlaced)
def handle_order(event: OrderPlaced):
    send_email(event.user_id, f"Order {event.order_id} confirmed")

This is the most Pythonic approach. Libraries like blinker and pyee use this pattern.

Class-Based Handlers

For complex handlers that need state or dependencies:

class InventoryHandler:
    def __init__(self, warehouse_client):
        self.warehouse = warehouse_client

    def handle(self, event: OrderPlaced):
        self.warehouse.reserve(event.order_id)

Register with bus.subscribe(OrderPlaced, inventory_handler.handle). This separates handler construction (where dependencies are injected) from handler registration (where the bus connects events to handlers).

Async Event Buses

For async Python applications, the bus should support coroutine handlers:

class AsyncEventBus:
    def __init__(self):
        self._handlers = defaultdict(list)

    def subscribe(self, event_type, handler):
        self._handlers[event_type].append(handler)

    async def publish(self, event):
        handlers = self._handlers[type(event)]
        await asyncio.gather(*(h(event) for h in handlers))

Using asyncio.gather runs all handlers concurrently. If one handler should not block others, wrap it in asyncio.create_task with error handling.

When to Use an Event Bus

Good fits:

• Multiple independent reactions to the same event (notifications, logging, cache updates).
• Plugin architectures where new behaviors are added without modifying existing code.
• Domain-driven design where domain events decouple aggregates.

Poor fits:

• Request-response patterns where the caller needs a result.
• Workflows with strict ordering requirements (event buses do not guarantee handler execution order by default).
• Small applications where direct function calls are clearer.

Event Bus vs Message Broker

An in-process event bus routes events within a single application. A message broker (Kafka, RabbitMQ, Redis Pub/Sub) routes events between separate applications or services. They solve the same conceptual problem at different scales.

For microservices, you often use both: an in-process event bus for internal decoupling within a service, and a message broker for communication between services.

Common Misconception

An event bus does not automatically make code easier to understand. Overuse creates “invisible wiring” where tracing what happens after an event is published requires searching the entire codebase for subscribers. Use event buses for genuine cross-cutting concerns, not as a replacement for direct function calls between tightly related components.

The one thing to remember: An event bus decouples producers from consumers inside a Python application — powerful for modularity, but overuse creates hidden dependencies that are harder to trace than direct calls.