How to Implement a Simple Event Bus in Python

from collections import defaultdict
from dataclasses import dataclass, field
from typing import Callable, Dict, List, Set


@dataclass
class EventBus:
    _subscribers: Dict[str, List[Callable]] = field(
        default_factory=lambda: defaultdict(list)
    )

    def subscribe(self, event_type: str, handler: Callable) -> None:
        self._subscribers[event_type].append(handler)

    def publish(self, event_type: str, payload: dict = None) -> None:
        for handler in self._subscribers.get(event_type, []):
            handler(payload or {})


def order_created_handler(data):
    print(f"Order created: {data['order_id']} for {data['customer']}")


def inventory_handler(data):
    print(f"Inventory reserved: {data['quantity']} units of {data['sku']}")


if __name__ == "__main__":
    bus = EventBus()
    bus.subscribe("order.created", order_created_handler)
    bus.subscribe("order.created", inventory_handler)

    bus.publish("order.created", {
        "order_id": 12345,
        "customer": "Alice",
        "sku": "BOOK-001",
        "quantity": 2
    })