Concepts / Behavioral patterns / Observer and event listeners

Observer and event listeners

Telling interested objects something happened, without knowing who they are.

Low-level design conceptBehavioral patterns2 Nov 2026

01The intuition

The problem it solves

A classroom app keeps the score of a quiz. When the score changes, the screen must redraw, a sound must play, and a log line must be written. The obvious code calls all three from inside the score class. Next week the teacher wants a leaderboard too, so the score class is edited again, and now it imports the screen, the speaker, the log and the leaderboard. A unit test of Score plays a sound.

class Score:
    def __init__(self) -> None:
        self.points = 0

    def add(self, n: int) -> None:
        self.points += n
        screen.redraw(self.points)
        speaker.play("ding.wav")
        log.write(f"score {self.points}")
        leaderboard.update(self.points)   # added this week, and next week...

Build it one piece at a time

Step one: turn the four calls into one idea, "something happened", and let the score keep a list of whoever wants to hear about it. The score does not know what they do with the news.

from typing import Callable

Listener = Callable[[int], None]

class Score:
    def __init__(self) -> None:
        self.points = 0
        self._listeners: list[Listener] = []

    def subscribe(self, fn: Listener) -> None:
        self._listeners.append(fn)

    def add(self, n: int) -> None:
        self.points += n
        for fn in list(self._listeners):   # copy: a listener may unsubscribe
            fn(self.points)

Step two: register the interested parts from the outside, at start-up. Each one is an observer; the score is the subject. Adding the leaderboard is one line in start-up code, and the Score test can subscribe a fake that just records calls.

score = Score()
score.subscribe(screen.redraw)
score.subscribe(lambda p: speaker.play("ding.wav"))
score.subscribe(lambda p: log.write(f"score {p}"))
score.subscribe(leaderboard.update)

seen: list[int] = []
test_score = Score(); test_score.subscribe(seen.append)
test_score.add(5); assert seen == [5]

Step three: pass an event object instead of a bare number once listeners need more than one fact, and give subscribe a way to undo itself so a closed screen stops being told.

from dataclasses import dataclass

@dataclass(frozen=True)
class ScoreChanged:
    old: int
    new: int

class Score:
    def subscribe(self, fn: Callable[[ScoreChanged], None]) -> Callable[[], None]:
        self._listeners.append(fn)
        return lambda: self._listeners.remove(fn)   # call to unsubscribe

Why it works

The principle: the thing that changes should not know who cares. Dependencies now point one way, from the listeners to the subject, so the subject stays small and testable. The invariant is that every subscribed listener is told about every change after it has happened, in subscription order, and nothing is told twice. Where it breaks: a listener that throws and stops the others from being notified, a listener that modifies the subject during notification and causes a second round, and listeners that are never removed and keep dead screens alive. Interviewers probe with "what if a listener is slow?" (notify on a queue or a thread, and accept that the order is no longer guaranteed) and "what if two threads add points at once?" (lock around the update and copy the list before iterating).

The trade-off you must say out loud

You lose the ability to read the program top to bottom: nothing in Score says a sound will play. Debugging means finding who subscribed. If there is exactly one listener and it will stay that way, a direct call is clearer. The usual follow-up is about delivery guarantees, which turns a plain observer into an event bus with queues, retries and ordering.

Where it shows up in the 100 Google problems

02Used in these problems