Observer Pattern

When a YouTube channel uploads a video, every subscriber is notified. The channel doesn't know who they are personally; it just keeps a list and tells everyone on it. Observer builds exactly that: a subject keeps a list of observers and notifies them all when its state changes, so new kinds of listeners can be added without touching the subject.

Intent#

Define a one-to-many dependency between objects so that when one object changes state, all its dependents are notified and updated automatically. (GoF)

Also known as publish-subscribe, dependents, or simply event listeners.

The problem: the subject knows everyone#

Without the pattern, the weather station updates each display by name:

class WeatherStation:
    def __init__(self) -> None:
        self.temperature = 0.0
        self.current_display = CurrentDisplay()
        self.stats_display = StatisticsDisplay()
        self.forecast_display = ForecastDisplay()

    def set_temperature(self, temp: float) -> None:
        self.temperature = temp
        # Every display is updated by hand
        self.current_display.update(temp)
        self.stats_display.update(temp)
        self.forecast_display.update(temp)

• The station is coupled to every concrete display.
• Adding or removing a display means editing WeatherStation.
• You can't add or remove displays at runtime.

Structure#

The subject keeps a list of observers and loops over it whenever its state changes.

The subject depends only on the Observer interface, never on concrete classes. That's the loose coupling.

Push vs pull#

• Push: the subject sends the data in the call: update(temperature, humidity) or update(videoTitle). Simple, but every observer receives everything.
• Pull: the subject sends only "something changed" (often passing itself), and each observer fetches what it needs: update(subject) → subject.getTemperature(). More flexible, but observers become coupled to the subject's getters.

Example 1: YouTube channel (push model)#

Here the channel pushes the video title to every subscriber.

from abc import ABC, abstractmethod


# Observer
class Subscriber(ABC):
    @abstractmethod
    def on_new_video(self, channel: str, video_title: str) -> None: ...


# Subject
class YoutubeChannel(ABC):
    @abstractmethod
    def add_subscriber(self, s: Subscriber) -> None: ...

    @abstractmethod
    def remove_subscriber(self, s: Subscriber) -> None: ...

    @abstractmethod
    def notify_subscribers(self, video_title: str) -> None: ...


# Concrete subject
class Channel(YoutubeChannel):
    def __init__(self, name: str) -> None:
        self.name = name
        self._subscribers: list[Subscriber] = []

    def add_subscriber(self, s: Subscriber) -> None:
        self._subscribers.append(s)

    def remove_subscriber(self, s: Subscriber) -> None:
        self._subscribers.remove(s)

    def post_video(self, title: str) -> None:
        print(f"{self.name} uploaded: {title}")
        self.notify_subscribers(title)  # state change triggers the broadcast

    def notify_subscribers(self, title: str) -> None:
        # Iterate over a copy: a subscriber may unsubscribe mid-loop
        for s in list(self._subscribers):
            s.on_new_video(self.name, title)


# Concrete observers
class EmailSubscriber(Subscriber):
    def __init__(self, email: str) -> None:
        self.email = email

    def on_new_video(self, channel: str, title: str) -> None:
        print(f"Email to {self.email}: {channel} posted '{title}'")


class PushSubscriber(Subscriber):
    def __init__(self, user: str) -> None:
        self.user = user

    def on_new_video(self, channel: str, title: str) -> None:
        print(f"Push to {self.user}: new video '{title}'")


alice = EmailSubscriber("alice@example.com")
bob = PushSubscriber("bob")

channel = Channel("CodingChannel")
channel.add_subscriber(alice)
channel.add_subscriber(bob)
channel.post_video("Design Patterns")  # both notified

channel.remove_subscriber(alice)
channel.post_video("System Design")  # only bob notified

Example 2: weather station (pull model)#

Here the station passes itself, and each observer reads the values it cares about.

from abc import ABC, abstractmethod


class Observer(ABC):
    @abstractmethod
    def update(self, subject: "WeatherStation") -> None: ...


class WeatherStation:
    def __init__(self) -> None:
        self._observers: list[Observer] = []
        self.temperature = 0.0
        self.humidity = 0.0

    def attach(self, o: Observer) -> None:
        self._observers.append(o)

    def detach(self, o: Observer) -> None:
        self._observers.remove(o)

    def set_measurements(
        self, temperature: float, humidity: float
    ) -> None:
        self.temperature, self.humidity = temperature, humidity
        # Copy the list: safe if an observer detaches itself
        for o in list(self._observers):
            o.update(self)  # pull model: observers read what they need


class CurrentDisplay(Observer):
    def update(self, subject: WeatherStation) -> None:
        print(f"Now: {subject.temperature}°C, {subject.humidity}% humidity")


class HeatAlert(Observer):
    def update(self, subject: WeatherStation) -> None:
        if subject.temperature > 35:
            print("Heat alert!")


station = WeatherStation()
station.attach(CurrentDisplay())
station.attach(HeatAlert())
station.set_measurements(38.0, 20.0)
# Now: 38.0°C, 20.0% humidity
# Heat alert!

Adding a forecast display or a mobile app is a new Observer class plus one attach() call. WeatherStation doesn't change.

Real-world examples#

You'll find Observer in almost every event system:

• GUI events: Tkinter's widget.bind(...) callbacks, DOM addEventListener in JavaScript.
• Python's logging: a logger passes every record to each attached handler.
• Node.js EventEmitter, .NET events and delegates.
• Reactive libraries: ReactiveX (RxPY, RxJS), Android LiveData (a lifecycle-aware observable).
• MVC/MVP/MVVM: the model notifies views when data changes.
• Domain examples: stock price tickers feeding several charts, chat rooms broadcasting messages, auction houses notifying bidders, social media notifications.

When to use it#

Reach for Observer when:

• A change to one object must update others, and you don't know in advance how many.
• Observers should be added and removed at runtime.
• The subject shouldn't know concrete observer classes (loose coupling).
• You're building event-driven code.

When not to use it#

Skip it when:

• There's exactly one dependent and it never changes. A direct call is clearer.
• Updates must happen in a strict order, or as one transaction.
• Updates are extremely frequent and observers are expensive (consider batching or throttling).

Pitfalls#

Most Observer bugs come from the notification loop itself.

• Memory leaks (the "lapsed listener" problem). The subject holds strong references to observers. If an observer is never detached, it can't be garbage-collected. Always unsubscribe when a component goes away.
• Modifying the list while notifying. An observer that unsubscribes inside update() can break iteration. Iterate over a copy.
• No ordering guarantee. Don't write observers that depend on running before or after each other.
• Update cascades. Observer A's update changes subject B, which notifies C, which... These chains are hard to debug.
• Slow or failing observers. In a simple loop, one slow observer delays all the others, and an exception can stop the loop. Catch errors per observer or notify asynchronously.

Observer vs similar patterns#

Observer is easy to confuse with these:

Key takeaways#

• The subject keeps a list of observers and calls update() on each when its state changes.
• The subject depends only on the observer interface, so new observers need no changes to it.
• Push sends data with the notification; pull lets observers fetch what they need.
• Unsubscribe to avoid leaks, iterate over a copy, and never rely on notification order.