How to Implement Observer Pattern in Python

Key Insights

The Observer pattern decouples state management from notification logic, allowing objects to subscribe to changes without tight coupling between components.
Python’s dynamic nature enables cleaner implementations using properties, descriptors, and weak references compared to traditional object-oriented languages.
Understanding when to use observers versus modern alternatives like asyncio event loops or reactive libraries prevents over-engineering simple notification systems.

The Observer pattern solves a fundamental problem in software design: how do you notify multiple objects about state changes without creating tight coupling? Think of it like a newsletter subscription—when new content is published, all subscribers automatically receive updates without the publisher needing to know who they are or what they do with the information.

This pattern is everywhere in modern applications: UI frameworks use it to update displays when data changes, event systems rely on it for decoupled communication, and reactive programming builds entire architectures around this concept. Let’s implement it properly in Python.

Core Components and Structure

The Observer pattern consists of four key elements:

Subject (Observable): The object being watched. It maintains a list of observers and notifies them of state changes.

Observer Interface: Defines the update method that all concrete observers must implement.

Concrete Subject: The actual object with state that changes. Inherits from Subject.

Concrete Observers: Objects that want to be notified. They implement the Observer interface and register with the subject.

Here’s the basic structure using Python’s Abstract Base Classes:

from abc import ABC, abstractmethod
from typing import List

class Observer(ABC):
    """Observer interface that all concrete observers must implement"""
    @abstractmethod
    def update(self, subject: 'Subject') -> None:
        pass

class Subject(ABC):
    """Subject interface that maintains and notifies observers"""
    def __init__(self):
        self._observers: List[Observer] = []
    
    def attach(self, observer: Observer) -> None:
        if observer not in self._observers:
            self._observers.append(observer)
    
    def detach(self, observer: Observer) -> None:
        self._observers.remove(observer)
    
    def notify(self) -> None:
        for observer in self._observers:
            observer.update(self)

Basic Implementation

Let’s build a weather station that broadcasts updates to multiple display devices. This is the classic example because it clearly demonstrates one-to-many relationships:

class WeatherStation(Subject):
    """Concrete subject that tracks weather data"""
    def __init__(self):
        super().__init__()
        self._temperature: float = 0.0
        self._humidity: float = 0.0
        self._pressure: float = 0.0
    
    def set_measurements(self, temp: float, humidity: float, pressure: float):
        self._temperature = temp
        self._humidity = humidity
        self._pressure = pressure
        self.notify()  # Notify all observers of the change
    
    @property
    def temperature(self) -> float:
        return self._temperature
    
    @property
    def humidity(self) -> float:
        return self._humidity
    
    @property
    def pressure(self) -> float:
        return self._pressure

class CurrentConditionsDisplay(Observer):
    """Concrete observer that displays current conditions"""
    def update(self, subject: Subject) -> None:
        if isinstance(subject, WeatherStation):
            print(f"Current conditions: {subject.temperature}°C, "
                  f"{subject.humidity}% humidity")

class StatisticsDisplay(Observer):
    """Concrete observer that tracks statistics"""
    def __init__(self):
        self._temperatures: List[float] = []
    
    def update(self, subject: Subject) -> None:
        if isinstance(subject, WeatherStation):
            self._temperatures.append(subject.temperature)
            avg_temp = sum(self._temperatures) / len(self._temperatures)
            print(f"Avg/Max/Min temperature: {avg_temp:.1f}/"
                  f"{max(self._temperatures):.1f}/"
                  f"{min(self._temperatures):.1f}")

# Usage
weather_station = WeatherStation()
current_display = CurrentConditionsDisplay()
stats_display = StatisticsDisplay()

weather_station.attach(current_display)
weather_station.attach(stats_display)

weather_station.set_measurements(25.5, 65.0, 1013.1)
weather_station.set_measurements(26.2, 70.0, 1012.8)

This produces:

Current conditions: 25.5°C, 65.0% humidity
Avg/Max/Min temperature: 25.5/25.5/25.5
Current conditions: 26.2°C, 70.0% humidity
Avg/Max/Min temperature: 25.8/26.2/25.5

Python-Specific Approaches

Python’s property decorators allow for more elegant implementations. Instead of manually calling notify(), we can trigger it automatically when state changes:

class SmartWeatherStation:
    """Weather station using properties for automatic notification"""
    def __init__(self):
        self._observers: List[Observer] = []
        self._temperature: float = 0.0
    
    @property
    def temperature(self) -> float:
        return self._temperature
    
    @temperature.setter
    def temperature(self, value: float) -> None:
        self._temperature = value
        self._notify_observers()
    
    def attach(self, observer: Observer) -> None:
        if observer not in self._observers:
            self._observers.append(observer)
    
    def _notify_observers(self) -> None:
        for observer in self._observers:
            observer.update(self)

# Usage is cleaner
station = SmartWeatherStation()
station.attach(current_display)
station.temperature = 27.3  # Automatically notifies observers

Real-World Use Case: Stock Price Monitoring

Let’s build a practical stock monitoring system where investors get notified when prices cross their thresholds:

from typing import Callable, Dict
from enum import Enum

class PriceChange(Enum):
    ABOVE_THRESHOLD = "above"
    BELOW_THRESHOLD = "below"

class StockTicker(Subject):
    """Tracks stock prices and notifies observers of significant changes"""
    def __init__(self, symbol: str):
        super().__init__()
        self.symbol = symbol
        self._price: float = 0.0
    
    def update_price(self, new_price: float) -> None:
        old_price = self._price
        self._price = new_price
        self.notify()
    
    @property
    def price(self) -> float:
        return self._price

class InvestorAlert(Observer):
    """Observer that alerts when price crosses thresholds"""
    def __init__(self, name: str, buy_threshold: float, sell_threshold: float):
        self.name = name
        self.buy_threshold = buy_threshold
        self.sell_threshold = sell_threshold
    
    def update(self, subject: Subject) -> None:
        if isinstance(subject, StockTicker):
            if subject.price <= self.buy_threshold:
                print(f"[{self.name}] BUY ALERT: {subject.symbol} "
                      f"at ${subject.price:.2f}")
            elif subject.price >= self.sell_threshold:
                print(f"[{self.name}] SELL ALERT: {subject.symbol} "
                      f"at ${subject.price:.2f}")

class TradingBot(Observer):
    """Automated trading observer"""
    def __init__(self, strategy: Callable[[float], str]):
        self.strategy = strategy
    
    def update(self, subject: Subject) -> None:
        if isinstance(subject, StockTicker):
            action = self.strategy(subject.price)
            if action:
                print(f"[BOT] Executing {action} for {subject.symbol}")

# Usage
tesla_stock = StockTicker("TSLA")

investor1 = InvestorAlert("Alice", buy_threshold=200.0, sell_threshold=250.0)
investor2 = InvestorAlert("Bob", buy_threshold=180.0, sell_threshold=270.0)

def simple_strategy(price: float) -> str:
    if price < 190:
        return "BUY"
    elif price > 260:
        return "SELL"
    return ""

bot = TradingBot(simple_strategy)

tesla_stock.attach(investor1)
tesla_stock.attach(investor2)
tesla_stock.attach(bot)

# Simulate price changes
tesla_stock.update_price(195.50)
tesla_stock.update_price(185.00)
tesla_stock.update_price(265.00)

Advanced Patterns and Considerations

Weak References: Prevent memory leaks by using weak references for observers. This allows observers to be garbage collected even if they’re still registered:

import weakref

class SafeSubject:
    """Subject using weak references to prevent memory leaks"""
    def __init__(self):
        self._observers: List[weakref.ref] = []
    
    def attach(self, observer: Observer) -> None:
        self._observers.append(weakref.ref(observer))
    
    def notify(self) -> None:
        # Clean up dead references and notify live ones
        self._observers = [obs for obs in self._observers if obs() is not None]
        for obs_ref in self._observers:
            observer = obs_ref()
            if observer:
                observer.update(self)

Thread Safety: For multi-threaded applications, protect the observer list:

import threading

class ThreadSafeSubject(Subject):
    def __init__(self):
        super().__init__()
        self._lock = threading.Lock()
    
    def attach(self, observer: Observer) -> None:
        with self._lock:
            super().attach(observer)
    
    def notify(self) -> None:
        with self._lock:
            observers = self._observers.copy()
        for observer in observers:
            observer.update(self)

Push vs. Pull Models: The examples above use a pull model (observers query the subject). For a push model, pass data directly:

class Observer(ABC):
    @abstractmethod
    def update(self, **kwargs) -> None:
        pass

# In subject
def notify(self) -> None:
    for observer in self._observers:
        observer.update(temperature=self._temperature, humidity=self._humidity)

Conclusion and Best Practices

Use the Observer pattern when:

Multiple objects need to react to state changes in another object
You want loose coupling between components
The number of observers is dynamic or unknown at compile time

Avoid it when:

You have only one or two observers (direct method calls are simpler)
You need complex event filtering (consider event buses or message queues)
You’re building async systems (use asyncio event loops instead)

Common pitfalls:

Memory leaks: Always use weak references or explicit detachment
Notification storms: Batch updates to avoid excessive notifications
Order dependency: Observers should be independent; if order matters, you need a different pattern

Modern Python alternatives include the asyncio library for async event handling, the blinker library for signals, and reactive frameworks like RxPY. For simple cases, Python’s built-in property decorators often suffice.

The Observer pattern remains valuable for understanding event-driven architecture, but evaluate whether simpler or more modern approaches fit your specific use case before implementing a full observer system.