Observer Pattern
Intent:
The Observer Pattern lets one object (subject) automatically notify many other objects (observers) when something changes. Instead of objects constantly checking for updates, the subject tells them when something happens.
When to Use:
Use the Observer Pattern when:
- One thing changes and many other things need to know about it
- You want objects to update automatically without checking constantly
- You want to easily add or remove things that need to be notified
- You want the notifier and the listeners to be independent
Key Characteristics:
- One subject (the thing that changes) notifies many observers (the things watching)
- Observers get updated automatically when the subject changes
- Subject and observers don't need to know much about each other
- Easy to add or remove observers anytime
Think of it like a weather station - when the temperature changes, it automatically notifies all TVs and phones showing the weather. They don't need to keep checking; they just get notified when something changes.
🌡️ Example: Weather Station Notifying Multiple Displays
Let's say we have a weather station that measures temperature. Multiple devices (TV, Mobile) want to show this temperature. When the temperature changes, all devices should be updated automatically.
Without Observer Pattern: Each device would need to constantly check the weather station for updates, which wastes time and resources. Or the weather station would need to know about every device, making it hard to add new devices.
With Observer Pattern: Devices register with the weather station to get updates. When temperature changes, the weather station automatically tells all registered devices. Devices can be added or removed easily without changing the weather station code.
Step 1: Define Observer Interface
# observer_interface.py from abc import ABC, abstractmethod class Observer(ABC): @abstractmethod def update(self, temperature): pass # All observers must implement this# observer_interface.py from abc import ABC, abstractmethod class Observer(ABC): @abstractmethod def update(self, temperature): pass # All observers must implement this
Step 2: Create Subject Interface
# subject_interface.py from abc import ABC, abstractmethod class Subject(ABC): @abstractmethod def register_observer(self, observer): pass # Add an observer @abstractmethod def remove_observer(self, observer): pass # Remove an observer @abstractmethod def notify_observers(self): pass # Notify all observers# subject_interface.py from abc import ABC, abstractmethod class Subject(ABC): @abstractmethod def register_observer(self, observer): pass # Add an observer @abstractmethod def remove_observer(self, observer): pass # Remove an observer @abstractmethod def notify_observers(self): pass # Notify all observers
Step 3: Create Weather Station (Subject)
# weather_station.py from subject_interface import Subject from observer_interface import Observer class WeatherStation(Subject): def __init__(self): self.observers = [] # List of observers self.temperature = 0 # Current temperature def register_observer(self, observer: Observer): """Add an observer to the list""" self.observers.append(observer) print(f"[Weather Station] Observer registered: {observer.__class__.__name__}") def remove_observer(self, observer: Observer): """Remove an observer from the list""" if observer in self.observers: self.observers.remove(observer) print(f"[Weather Station] Observer removed: {observer.__class__.__name__}") def notify_observers(self): """Notify all observers about temperature change""" print(f"\n[Weather Station] Notifying {len(self.observers)} observers...") for observer in self.observers: observer.update(self.temperature) def set_temperature(self, temperature): """Update temperature and notify all observers""" print(f"\n[Weather Station] Temperature changed: {self.temperature}°C → {temperature}°C") self.temperature = temperature self.notify_observers() # Automatically notify all observers# weather_station.py from subject_interface import Subject from observer_interface import Observer class WeatherStation(Subject): def __init__(self): self.observers = [] # List of observers self.temperature = 0 # Current temperature def register_observer(self, observer: Observer): """Add an observer to the list""" self.observers.append(observer) print(f"[Weather Station] Observer registered: {observer.__class__.__name__}") def remove_observer(self, observer: Observer): """Remove an observer from the list""" if observer in self.observers: self.observers.remove(observer) print(f"[Weather Station] Observer removed: {observer.__class__.__name__}") def notify_observers(self): """Notify all observers about temperature change""" print(f"\n[Weather Station] Notifying {len(self.observers)} observers...") for observer in self.observers: observer.update(self.temperature) def set_temperature(self, temperature): """Update temperature and notify all observers""" print(f"\n[Weather Station] Temperature changed: {self.temperature}°C → {temperature}°C") self.temperature = temperature self.notify_observers() # Automatically notify all observers
Step 4: Create TV Display (Observer)
# tv_display.py from observer_interface import Observer class TVDisplay(Observer): def __init__(self, name): self.name = name self.current_temperature = None def update(self, temperature): """Called when weather station notifies about temperature change""" self.current_temperature = temperature print(f" 📺 [{self.name}] TV Display updated: {temperature}°C")# tv_display.py from observer_interface import Observer class TVDisplay(Observer): def __init__(self, name): self.name = name self.current_temperature = None def update(self, temperature): """Called when weather station notifies about temperature change""" self.current_temperature = temperature print(f" 📺 [{self.name}] TV Display updated: {temperature}°C")
Step 5: Create Mobile Display (Observer)
# mobile_display.py from observer_interface import Observer class MobileDisplay(Observer): def __init__(self, name): self.name = name self.current_temperature = None def update(self, temperature): """Called when weather station notifies about temperature change""" self.current_temperature = temperature print(f" 📱 [{self.name}] Mobile Display updated: {temperature}°C")# mobile_display.py from observer_interface import Observer class MobileDisplay(Observer): def __init__(self, name): self.name = name self.current_temperature = None def update(self, temperature): """Called when weather station notifies about temperature change""" self.current_temperature = temperature print(f" 📱 [{self.name}] Mobile Display updated: {temperature}°C")
Step 6: Use the Observer Pattern (Client Code)
# main.py from weather_station import WeatherStation from tv_display import TVDisplay from mobile_display import MobileDisplay # Create weather station (subject) weather_station = WeatherStation() # Create observers (displays) living_room_tv = TVDisplay("Living Room TV") my_mobile = MobileDisplay("My Mobile") # Register observers with weather station print("=== Registering Observers ===") weather_station.register_observer(living_room_tv) weather_station.register_observer(my_mobile) # Change temperature - all observers are automatically notified! print("\n=== Temperature Updates ===") weather_station.set_temperature(25) weather_station.set_temperature(28) weather_station.set_temperature(22) # Remove an observer print("\n=== Removing Observer ===") weather_station.remove_observer(living_room_tv) # Change temperature again - only remaining observers are notified print("\n=== Temperature Update After Removal ===") weather_station.set_temperature(20)# main.py from weather_station import WeatherStation from tv_display import TVDisplay from mobile_display import MobileDisplay # Create weather station (subject) weather_station = WeatherStation() # Create observers (displays) living_room_tv = TVDisplay("Living Room TV") my_mobile = MobileDisplay("My Mobile") # Register observers with weather station print("=== Registering Observers ===") weather_station.register_observer(living_room_tv) weather_station.register_observer(my_mobile) # Change temperature - all observers are automatically notified! print("\n=== Temperature Updates ===") weather_station.set_temperature(25) weather_station.set_temperature(28) weather_station.set_temperature(22) # Remove an observer print("\n=== Removing Observer ===") weather_station.remove_observer(living_room_tv) # Change temperature again - only remaining observers are notified print("\n=== Temperature Update After Removal ===") weather_station.set_temperature(20)
Output:
=== Registering Observers === [Weather Station] Observer registered: TVDisplay [Weather Station] Observer registered: MobileDisplay === Temperature Updates === [Weather Station] Temperature changed: 0°C → 25°C [Weather Station] Notifying 2 observers... 📺 [Living Room TV] TV Display updated: 25°C 📱 [My Mobile] Mobile Display updated: 25°C [Weather Station] Temperature changed: 25°C → 28°C [Weather Station] Notifying 2 observers... 📺 [Living Room TV] TV Display updated: 28°C 📱 [My Mobile] Mobile Display updated: 28°C [Weather Station] Temperature changed: 28°C → 22°C [Weather Station] Notifying 2 observers... 📺 [Living Room TV] TV Display updated: 22°C 📱 [My Mobile] Mobile Display updated: 22°C === Removing Observer === [Weather Station] Observer removed: TVDisplay === Temperature Update After Removal === [Weather Station] Temperature changed: 22°C → 20°C [Weather Station] Notifying 1 observers... 📱 [My Mobile] Mobile Display updated: 20°C
✅ Benefits
1. Automatic Updates:
- Observers get notified automatically when the subject changes
- No need to keep checking for updates - saves time and resources
2. Easy to Add/Remove:
- The subject doesn't need to know details about observers
- You can add or remove observers easily without changing other code
3. Easy to Extend:
- Want to add a new observer? Just create it and register it
- Don't need to change the subject or other observers
4. Efficient Communication:
- One change automatically tells many observers at once
- Perfect for systems where one thing affects many others
🎯 Key Points
- Observer Pattern lets one thing notify many things automatically
- When the subject changes, it tells all registered observers
- Observers sign up (register) to get updates from the subject
- Easy to add or remove observers anytime
- Perfect for systems where one change affects many things
🌟 Quick Summary
- Problem: You need to tell many objects when one object changes, but you don't want them constantly checking or tightly connected
- Solution: The subject keeps a list of observers and automatically tells them all when something changes
- Benefit: Objects get updated automatically without being tightly connected, and it's easy to add or remove observers
- Example: Weather station automatically tells TV and Mobile when temperature changes - they don't need to keep checking
🎉 End
The Observer Pattern helps you create systems where objects automatically stay updated. It's perfect when one thing changes and many other things need to know about it, like notifications, updates, or event systems!