The Observer Pattern is a behavioral design pattern that defines a one-to-many dependency between objects. When one object (the subject) changes state, all its dependents (observers) are automatically notified and updated. This pattern promotes loose coupling between objects by ensuring that the subject doesn’t need to know the concrete classes of its observers.
Core Concepts
The Observer Pattern consists of two main components:
Subject (Observable): The object that maintains a list of observers and notifies them of state changes. It provides methods to attach, detach, and notify observers.
Observer: An interface or abstract class that defines the update method that concrete observers must implement. Concrete observers register themselves with subjects they’re interested in monitoring.
Pattern Structure
┌─────────────────┐ ┌──────────────────┐
│ Subject │◊─────▶│ Observer │
├─────────────────┤ ├──────────────────┤
│ +attach() │ │ +update() │
│ +detach() │ └──────────────────┤
│ +notify() │ ▲ │
└─────────────────┘ │ │
▲ │ │
│ │ │
┌─────────────────┐ ┌──────────────────┐
│ ConcreteSubject │ │ ConcreteObserver │
├─────────────────┤ ├──────────────────┤
│ -state │ │ -subject_state │
│ +get_state() │ │ +update() │
│ +set_state() │ └──────────────────┘
└─────────────────┘
When to Use the Observer Pattern
The Observer Pattern is particularly useful when:
- You need to maintain consistency between related objects without making them tightly coupled
- A change in one object requires updating multiple dependent objects
- You want to broadcast information to multiple recipients without knowing who they are
- You need to support undo operations or event logging
- You’re implementing model-view architectures where views need to stay synchronized with model changes
Benefits and Trade-offs
Benefits:
- Promotes loose coupling between subjects and observers
- Supports broadcast communication
- Allows dynamic relationships between objects
- Follows the Open/Closed Principle - you can add new observers without modifying existing code
Trade-offs:
- Can lead to memory leaks if observers aren’t properly detached
- Notification order is not guaranteed
- Complex update dependencies can cause unexpected cascading effects
- Debugging can be challenging due to indirect relationships
Example 1: Weather Station Monitoring System
This example demonstrates a weather monitoring system where a weather station (subject) notifies multiple display devices (observers) whenever weather conditions change. Each display shows different aspects of the weather data.
from abc import ABC, abstractmethod
from typing import List, Protocol
class Observer(Protocol):
"""Observer interface that defines the update contract."""
def update(self, temperature: float, humidity: float, pressure: float) -> None:
"""Called when the subject's state changes."""
...
class Subject(ABC):
"""Abstract subject class that manages observers."""
def __init__(self) -> None:
self._observers: List[Observer] = []
def attach(self, observer: Observer) -> None:
"""Attach an observer to the subject."""
if observer not in self._observers:
self._observers.append(observer)
print(f"Observer {observer.__class__.__name__} attached")
def detach(self, observer: Observer) -> None:
"""Detach an observer from the subject."""
if observer in self._observers:
self._observers.remove(observer)
print(f"Observer {observer.__class__.__name__} detached")
def notify(self) -> None:
"""Notify all attached observers of state change."""
print("Notifying observers...")
for observer in self._observers:
observer.update(*self.get_measurements())
@abstractmethod
def get_measurements(self) -> tuple[float, float, float]:
"""Get current measurements to pass to observers."""
pass
class WeatherStation(Subject):
"""Concrete subject that represents a weather monitoring station."""
def __init__(self) -> None:
super().__init__()
self._temperature: float = 0.0
self._humidity: float = 0.0
self._pressure: float = 0.0
def get_measurements(self) -> tuple[float, float, float]:
"""Return current weather measurements."""
return self._temperature, self._humidity, self._pressure
def set_measurements(self, temperature: float, humidity: float, pressure: float) -> None:
"""Update weather measurements and notify observers."""
print(f"\nWeatherStation: New measurements received")
print(f"Temperature: {temperature}°C, Humidity: {humidity}%, Pressure: {pressure}hPa")
self._temperature = temperature
self._humidity = humidity
self._pressure = pressure
# Notify observers of the change
self.notify()
class CurrentConditionsDisplay:
"""Concrete observer that displays current weather conditions."""
def update(self, temperature: float, humidity: float, pressure: float) -> None:
"""Update display with new weather data."""
print(f"Current Conditions Display: {temperature}°C, {humidity}% humidity, {pressure}hPa")
class StatisticsDisplay:
"""Concrete observer that tracks weather statistics."""
def __init__(self) -> None:
self._temperature_readings: List[float] = []
self._max_temp: float = float('-inf')
self._min_temp: float = float('inf')
def update(self, temperature: float, humidity: float, pressure: float) -> None:
"""Update statistics with new temperature reading."""
self._temperature_readings.append(temperature)
self._max_temp = max(self._max_temp, temperature)
self._min_temp = min(self._min_temp, temperature)
avg_temp = sum(self._temperature_readings) / len(self._temperature_readings)
print(f"Statistics Display: Avg: {avg_temp:.1f}°C, Max: {self._max_temp}°C, Min: {self._min_temp}°C")
class ForecastDisplay:
"""Concrete observer that provides weather forecast based on pressure trends."""
def __init__(self) -> None:
self._last_pressure: float = 0.0
def update(self, temperature: float, humidity: float, pressure: float) -> None:
"""Generate forecast based on pressure changes."""
if self._last_pressure == 0.0:
forecast = "More weather data needed"
elif pressure > self._last_pressure:
forecast = "Improving weather on the way!"
elif pressure < self._last_pressure:
forecast = "Watch out for cooler, rainy weather"
else:
forecast = "More of the same"
print(f"Forecast Display: {forecast}")
self._last_pressure = pressure
# Example usage
if __name__ == "__main__":
# Create weather station (subject)
weather_station = WeatherStation()
# Create displays (observers)
current_display = CurrentConditionsDisplay()
statistics_display = StatisticsDisplay()
forecast_display = ForecastDisplay()
# Register observers with the subject
weather_station.attach(current_display)
weather_station.attach(statistics_display)
weather_station.attach(forecast_display)
print("=" * 50)
# Simulate weather updates
weather_station.set_measurements(25.5, 65.0, 1013.2)
print("\n" + "=" * 50)
weather_station.set_measurements(27.8, 58.0, 1015.1)
print("\n" + "=" * 50)
# Remove one observer
weather_station.detach(forecast_display)
weather_station.set_measurements(23.1, 72.0, 1008.9)
Key Features:
- WeatherStation acts as the subject that maintains weather measurements
- Multiple Display Classes serve as concrete observers, each with different display logic
- Protocol-based Observer Interface ensures type safety and clear contracts
- Automatic Notification happens whenever measurements are updated
The weather station doesn’t need to know what types of displays are connected or how they process the data. New display types can be added without modifying existing code, and displays can be connected or disconnected at runtime.
Example 2: Event-Driven System Architecture
This example demonstrates an event-driven architecture where different services publish events, and various observers handle these events for different purposes like notifications, logging, and analytics.
from abc import ABC, abstractmethod
from typing import Dict, List, Any, Callable
from enum import Enum
class EventType(Enum):
"""Enumeration of different event types in the system."""
USER_LOGIN = "user_login"
USER_LOGOUT = "user_logout"
ORDER_PLACED = "order_placed"
PAYMENT_PROCESSED = "payment_processed"
class Event:
"""Event data container that carries information about what happened."""
def __init__(self, event_type: EventType, data: Dict[str, Any]) -> None:
self.event_type = event_type
self.data = data
self.timestamp = self._get_current_time()
def _get_current_time(self) -> str:
"""Get current timestamp (simplified for example)."""
import datetime
return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
class EventObserver(ABC):
"""Abstract base class for event observers."""
@abstractmethod
def handle_event(self, event: Event) -> None:
"""Handle the received event."""
pass
class EventPublisher:
"""Event publisher that manages observers and distributes events."""
def __init__(self) -> None:
# Dictionary mapping event types to lists of observers
self._observers: Dict[EventType, List[EventObserver]] = {}
def subscribe(self, event_type: EventType, observer: EventObserver) -> None:
"""Subscribe an observer to a specific event type."""
if event_type not in self._observers:
self._observers[event_type] = []
if observer not in self._observers[event_type]:
self._observers[event_type].append(observer)
print(f"{observer.__class__.__name__} subscribed to {event_type.value} events")
def unsubscribe(self, event_type: EventType, observer: EventObserver) -> None:
"""Unsubscribe an observer from a specific event type."""
if event_type in self._observers and observer in self._observers[event_type]:
self._observers[event_type].remove(observer)
print(f"{observer.__class__.__name__} unsubscribed from {event_type.value} events")
def publish_event(self, event: Event) -> None:
"""Publish an event to all subscribed observers."""
print(f"\n📢 Publishing {event.event_type.value} event at {event.timestamp}")
if event.event_type in self._observers:
for observer in self._observers[event.event_type]:
observer.handle_event(event)
else:
print(f"No observers subscribed to {event.event_type.value}")
class EmailNotificationService(EventObserver):
"""Observer that sends email notifications for specific events."""
def handle_event(self, event: Event) -> None:
"""Send email notification based on event type."""
if event.event_type == EventType.USER_LOGIN:
user_email = event.data.get('email', 'unknown')
print(f"📧 EmailService: Sending welcome email to {user_email}")
elif event.event_type == EventType.ORDER_PLACED:
order_id = event.data.get('order_id', 'unknown')
user_email = event.data.get('user_email', 'unknown')
print(f"📧 EmailService: Sending order confirmation for #{order_id} to {user_email}")
class AuditLogger(EventObserver):
"""Observer that logs all events for audit purposes."""
def __init__(self) -> None:
self._log_entries: List[str] = []
def handle_event(self, event: Event) -> None:
"""Log event details for audit trail."""
log_entry = f"[{event.timestamp}] {event.event_type.value}: {event.data}"
self._log_entries.append(log_entry)
print(f"📋 AuditLogger: {log_entry}")
def get_logs(self) -> List[str]:
"""Return all logged entries."""
return self._log_entries.copy()
class AnalyticsTracker(EventObserver):
"""Observer that tracks events for analytics purposes."""
def __init__(self) -> None:
self._event_counts: Dict[EventType, int] = {}
def handle_event(self, event: Event) -> None:
"""Track event for analytics."""
if event.event_type not in self._event_counts:
self._event_counts[event.event_type] = 0
self._event_counts[event.event_type] += 1
print(f"📊 Analytics: {event.event_type.value} count: {self._event_counts[event.event_type]}")
# Special analytics for login events
if event.event_type == EventType.USER_LOGIN:
user_type = event.data.get('user_type', 'regular')
print(f"📊 Analytics: User type '{user_type}' logged in")
def get_statistics(self) -> Dict[EventType, int]:
"""Get current event statistics."""
return self._event_counts.copy()
class UserService:
"""Service that publishes user-related events."""
def __init__(self, event_publisher: EventPublisher) -> None:
self._event_publisher = event_publisher
def login_user(self, user_id: str, email: str, user_type: str = "regular") -> None:
"""Handle user login and publish login event."""
print(f"UserService: User {user_id} logging in...")
# Simulate login logic here
# Publish login event
login_event = Event(
EventType.USER_LOGIN,
{
'user_id': user_id,
'email': email,
'user_type': user_type
}
)
self._event_publisher.publish_event(login_event)
def logout_user(self, user_id: str) -> None:
"""Handle user logout and publish logout event."""
print(f"UserService: User {user_id} logging out...")
logout_event = Event(
EventType.USER_LOGOUT,
{'user_id': user_id}
)
self._event_publisher.publish_event(logout_event)
class OrderService:
"""Service that publishes order-related events."""
def __init__(self, event_publisher: EventPublisher) -> None:
self._event_publisher = event_publisher
def place_order(self, order_id: str, user_email: str, amount: float) -> None:
"""Process order and publish order event."""
print(f"OrderService: Processing order {order_id}...")
# Simulate order processing logic
# Publish order placed event
order_event = Event(
EventType.ORDER_PLACED,
{
'order_id': order_id,
'user_email': user_email,
'amount': amount
}
)
self._event_publisher.publish_event(order_event)
# Example usage
if __name__ == "__main__":
# Create event publisher (central hub)
event_publisher = EventPublisher()
# Create observers (event handlers)
email_service = EmailNotificationService()
audit_logger = AuditLogger()
analytics_tracker = AnalyticsTracker()
# Subscribe observers to specific event types
event_publisher.subscribe(EventType.USER_LOGIN, email_service)
event_publisher.subscribe(EventType.USER_LOGIN, audit_logger)
event_publisher.subscribe(EventType.USER_LOGIN, analytics_tracker)
event_publisher.subscribe(EventType.ORDER_PLACED, email_service)
event_publisher.subscribe(EventType.ORDER_PLACED, audit_logger)
event_publisher.subscribe(EventType.ORDER_PLACED, analytics_tracker)
# Only audit logger cares about logout events
event_publisher.subscribe(EventType.USER_LOGOUT, audit_logger)
print("=" * 60)
# Create services that publish events
user_service = UserService(event_publisher)
order_service = OrderService(event_publisher)
# Simulate system operations
user_service.login_user("user123", "john@example.com", "premium")
print("\n" + "=" * 60)
order_service.place_order("ORD-001", "john@example.com", 99.99)
print("\n" + "=" * 60)
user_service.logout_user("user123")
print("\n" + "=" * 60)
# Show analytics summary
print("\n📊 Final Analytics Summary:")
stats = analytics_tracker.get_statistics()
for event_type, count in stats.items():
print(f" {event_type.value}: {count} events")
Key Features:
- Event Publisher acts as a central hub that manages subscriptions and distributes events
- Event-Type-Specific Subscriptions allow observers to only receive events they care about
- Decoupled Services can publish events without knowing who will handle them
- Multiple Event Handlers can process the same event for different purposes (email, logging, analytics)
This pattern is particularly useful in microservices architectures, where different components need to react to system events without being tightly coupled. Services can focus on their core functionality while publishing events, and other components can subscribe to relevant events to perform their specific tasks.
The Observer Pattern promotes maintainable, scalable systems by establishing clear communication protocols between components while maintaining loose coupling. Whether used for UI updates, event-driven architectures, or data synchronization, this pattern provides a robust foundation for building responsive and modular applications.