An Event Bus

# labs/event_bus.py
from collections import defaultdict
from typing import Callable

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

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

    def publish(self, event: str, data=None) -> None:
        for handler in self._handlers[event]:
            handler(data)

bus = EventBus()
bus.subscribe("speech_detected", lambda d: print(f"[EVENT] speech_detected: {d}"))
bus.publish("speech_detected", {"text": "Hello GLaDOS"})

$ uv run python labs/event_bus.py
[EVENT] speech_detected: {'text': 'Hello GLaDOS'}

The entire mechanism is a dictionary from names to lists of functions. defaultdict(list) earns its import: the first subscriber for an event arrives before that event's list exists, and the defaultdict conjures an empty list on first touch, so neither method ever guards for a missing key. The other load-bearing fact is one you proved in chapter 2 without noticing: functions are objects. They sit in lists, and calling handler(data) later works because a function in a list is still a function.

    def unsubscribe(self, event: str, handler: Callable) -> None:
        self._handlers[event] = [
            h for h in self._handlers[event] if h is not handler
        ]

$ uv run python labs/event_bus.py
[EVENT] speech_detected: {'text': 'first'}
done

The demo subscribes a named handler, publishes once, unsubscribes, publishes again: one line of output, then silence, then "done." Two details are doing quiet work. Removal compares by identity (is not, never !=), because you want the exact function object the caller handed you gone, not anything that happens to compare equal. And the list is rebuilt with a comprehension instead of mutated in place, which stays safe even if a handler unsubscribes itself mid-publish. Note what the second publish did about having no audience: nothing. Hold that thought for the failure box.

bus = EventBus()   # module-level: every importer gets this one

def on_speech_detected(data):
    print(f"[EVENT] speech_detected: {data}")

def on_response_ready(data):
    print(f"[EVENT] response_ready: {data}")

if __name__ == "__main__":
    bus.subscribe("speech_detected", on_speech_detected)
    bus.subscribe("response_ready", on_response_ready)
    bus.publish("speech_detected", {"text": "Hello GLaDOS"})
    bus.publish("response_ready", {"reply": "Fascinating. You can speak."})

$ uv run python labs/event_bus.py
[EVENT] speech_detected: {'text': 'Hello GLaDOS'}
[EVENT] response_ready: {'reply': 'Fascinating. You can speak.'}

The module-level instance is the meeting point: every file that writes from event_bus import bus gets the same object, which is how the wake module and the mouth module coordinate without either importing the other. Their entire shared vocabulary is a set of strings. That is the decoupling, and it is also the danger, because strings have no compiler checking them. The __main__ guard keeps the demo wiring from firing on import, a habit you now apply without being told.