How to Implement the Observer Design Pattern in Python

Learn how to implement the Observer pattern in Python with a step-by-step guide covering subjects, observers, dynamic subscriptions, and practical use cases like event systems and real-time data feeds.

The Observer pattern is a behavioral design pattern that establishes a one-to-many dependency between objects. When one object (the subject or publisher) changes state, all its dependent objects (the observers or subscribers) are automatically notified and updated. This pattern is fundamental in event-driven programming and is used extensively in GUI frameworks, notification systems, real-time data feeds, and reactive architectures.

In this guide, you will learn how the Observer pattern works, implement it from scratch in Python, understand its advantages and trade-offs, and see practical examples that demonstrate when and why to use it.

The Problem: Tight Coupling and Unnecessary Polling

Imagine you are building an application that processes data and displays it in multiple formats: decimal, hexadecimal, and octal. Without the Observer pattern, you face two poor options:

Polling: Each viewer repeatedly checks the data source for changes, which is wasteful and inefficient.
Direct notification: The data source directly calls each viewer. However, the data source must then know about every viewer, creating tight coupling that makes it hard to add or remove viewers.

Both approaches violate the Open/Closed Principle (open for extension, closed for modification) because adding a new viewer requires modifying the data source.

The Solution: Observer Pattern

The Observer pattern solves this by introducing a subscription mechanism:

The Subject maintains a list of observers and provides methods to attach, detach, and notify them.
Observers register with the subject and implement an update() method that gets called whenever the subject's state changes.
The subject and observers are loosely coupled: the subject only knows that observers have an update() method, not what they do with the data.

Implementation

Step 1: Create the Subject Base Class

The Subject class manages the observer list and handles notification:

class Subject:
    """Base class for objects that need to be observed."""

    def __init__(self):
        self._observers = []

    def attach(self, observer):
        """Subscribe an observer to receive notifications."""
        if observer not in self._observers:
            self._observers.append(observer)

    def detach(self, observer):
        """Unsubscribe an observer from notifications."""
        try:
            self._observers.remove(observer)
        except ValueError:
            pass

    def notify(self, modifier=None):
        """Notify all observers about a state change.

        Args:
            modifier: The observer that triggered the change (optional).
                      This observer will be skipped to avoid infinite loops.
        """
        for observer in self._observers:
            if modifier != observer:
                observer.update(self)

Step 2: Create a Concrete Subject

The Data class extends Subject and automatically notifies observers whenever its data property changes:

class Data(Subject):
    """A monitored data source that notifies observers on change."""

    def __init__(self, name=''):
        super().__init__()
        self.name = name
        self._data = 0

    @property
    def data(self):
        return self._data

    @data.setter
    def data(self, value):
        self._data = value
        self.notify()  # Automatically notify all observers

Using a property setter ensures that every data change triggers notification; observers never miss an update.

Step 3: Create Observer Classes

Each observer implements an update() method that defines how it reacts to changes:

class DecimalViewer:
    """Displays data in decimal format."""

    def update(self, subject):
        print(f"DecimalViewer: Subject '{subject.name}' has data {subject.data}")


class HexViewer:
    """Displays data in hexadecimal format."""

    def update(self, subject):
        print(f"HexViewer: Subject '{subject.name}' has data 0x{subject.data:x}")


class OctalViewer:
    """Displays data in octal format."""

    def update(self, subject):
        print(f"OctalViewer: Subject '{subject.name}' has data {oct(subject.data)}")

Step 4: Wire Everything Together

if __name__ == "__main__":
    obj1 = Data('Data 1')
    obj2 = Data('Data 2')

    decimal_viewer = DecimalViewer()
    hex_viewer = HexViewer()
    octal_viewer = OctalViewer()

    obj1.attach(decimal_viewer)
    obj1.attach(hex_viewer)
    obj1.attach(octal_viewer)

    obj2.attach(decimal_viewer)
    obj2.attach(hex_viewer)
    obj2.attach(octal_viewer)

    obj1.data = 10
    print()
    obj2.data = 15

Output:

DecimalViewer: Subject 'Data 1' has data 10
HexViewer: Subject 'Data 1' has data 0xa
OctalViewer: Subject 'Data 1' has data 0o12

DecimalViewer: Subject 'Data 2' has data 15
HexViewer: Subject 'Data 2' has data 0xf
OctalViewer: Subject 'Data 2' has data 0o17

Every time data is set, all attached observers are automatically notified and display the value in their respective formats.

Demonstrating Dynamic Subscription

One of the key benefits of the Observer pattern is that observers can be added or removed at runtime:

if __name__ == "__main__":
    data = Data('Sensor')

    decimal_viewer = DecimalViewer()
    hex_viewer = HexViewer()

    data.attach(decimal_viewer)
    data.data = 42
    print()

    data.attach(hex_viewer)
    data.data = 255
    print()

    data.detach(decimal_viewer)
    data.data = 128

Output:

DecimalViewer: Subject 'Sensor' has data 42

DecimalViewer: Subject 'Sensor' has data 255
HexViewer: Subject 'Sensor' has data 0xff

HexViewer: Subject 'Sensor' has data 0x80

Observers are added and removed without modifying the Data class at all.

A More Pythonic Implementation Using Protocols

Python does not require formal interfaces, but using an abstract base class or a Protocol makes the observer contract explicit:

from abc import ABC, abstractmethod


class Observer(ABC):
    """Abstract observer that all concrete observers must implement."""

    @abstractmethod
    def update(self, subject):
        pass


class Subject:
    def __init__(self):
        self._observers: list[Observer] = []

    def attach(self, observer: Observer):
        if observer not in self._observers:
            self._observers.append(observer)

    def detach(self, observer: Observer):
        self._observers.remove(observer)

    def notify(self):
        for observer in self._observers:
            observer.update(self)

Now any observer must implement update(), and your IDE can catch missing implementations at development time.

Practical Example: Event-Driven Notification System

class EventManager:
    """A generic event system supporting multiple event types."""

    def __init__(self):
        self._listeners = {}

    def subscribe(self, event_type, listener):
        if event_type not in self._listeners:
            self._listeners[event_type] = []
        self._listeners[event_type].append(listener)

    def unsubscribe(self, event_type, listener):
        if event_type in self._listeners:
            self._listeners[event_type].remove(listener)

    def notify(self, event_type, data=None):
        for listener in self._listeners.get(event_type, []):
            listener(event_type, data)


def email_alert(event, data):
    print(f"[EMAIL] Event '{event}': {data}")

def log_event(event, data):
    print(f"[LOG] Event '{event}': {data}")

def sms_alert(event, data):
    print(f"[SMS] Event '{event}': {data}")


manager = EventManager()
manager.subscribe("user_signup", email_alert)
manager.subscribe("user_signup", log_event)
manager.subscribe("payment", sms_alert)
manager.subscribe("payment", log_event)

manager.notify("user_signup", {"user": "Alice"})
print()
manager.notify("payment", {"amount": 99.99})

Output:

[EMAIL] Event 'user_signup': {'user': 'Alice'}
[LOG] Event 'user_signup': {'user': 'Alice'}

[SMS] Event 'payment': {'amount': 99.99}
[LOG] Event 'payment': {'amount': 99.99}

This event-based variant supports multiple event types, each with its own set of subscribers.

Advantages and Disadvantages

Advantages

Benefit	Description
Open/Closed Principle	New observers can be added without modifying the subject
Loose coupling	Subject and observers interact through a minimal interface (`update()`)
Runtime flexibility	Observers can be attached and detached dynamically
One-to-many communication	One state change notifies any number of observers

Disadvantages

Drawback	Description
Memory leaks	Observers that are not properly detached can prevent garbage collection (the "lapsed listener" problem)
Unpredictable notification order	Observers are notified in registration order, which may not match business requirements
Cascade updates	An observer's `update()` might trigger further state changes, leading to complex chains
Debugging difficulty	Tracking which observer caused a particular behavior can be challenging

Avoid the lapsed listener problem

Always detach observers when they are no longer needed. In Python, consider using weak references (weakref) to allow observers to be garbage-collected even if they forget to unsubscribe:

import weakref

class Subject:
    def __init__(self):
        self._observers = weakref.WeakSet()

    def attach(self, observer):
        self._observers.add(observer)

    def notify(self):
        for observer in self._observers:
            observer.update(self)

With WeakSet, observers are automatically removed when no other references to them exist.

When to Use the Observer Pattern

Scenario	Example
Event systems	GUI button clicks, keyboard events
Data binding	UI automatically reflects model changes (MVC/MVVM)
Notification services	Email, SMS, push notifications on state changes
Real-time feeds	Stock tickers, social media updates, RSS feeds
Logging and monitoring	Multiple loggers observing application events
Plugin architectures	Plugins subscribe to application events

Conclusion

The Observer design pattern provides a clean, loosely coupled way to implement one-to-many communication between objects in Python.

The subject maintains a list of observers and notifies them automatically when its state changes, without needing to know the specifics of each observer.
This pattern is foundational to event-driven programming and is used extensively in frameworks, notification systems, and reactive architectures.

By using abstract base classes for the observer interface and weak references to prevent memory leaks, you can build robust, maintainable implementations that follow the Open/Closed Principle.