Observer

Observer

An Observer defines a one-to-many dependency between objects such that a change in state of one object notifies and updates all dependent objects. In other words, this works as a subscription mechanism for client objects, such that they can be notified of changes in the observed object.

TL;DR

Defines a dependency between objects so that when an object changes state, all dependent objects are notified.

Problem

Suppose you are creating a news agency application that needs to notify clients (mobile apps, websites, etc.) of breaking news. One way to do this would be to add a list of clients to the application and simply loop over them, but you might run into the issue where they are hard coded or difficult to manage - how do you add new ones, how do you remove clients that want to unsubscribe? Note that this is a particular issue when you want to manage this at runtime.

Solution

The solution is to create an Observer pattern. This consists of an observer interface and a subject (the subject is the observed object). The subject keeps track of all subscribed clients, and implements methods to add or remove them, as well as notify them. Meanwhile, the observers are concrete implementations of the Observer interface and have logic to handle specific updates.

from abc import ABC, abstractmethod
from typing import List
 
# Observer interface
class Observer(ABC):
    @abstractmethod
    def update(self, message: str):
        pass
 
# Subject (Observable) class
class NewsAgency:
    def __init__(self):
        self._observers: List[Observer] = []
        self._news = ""
    
    def add_observer(self, observer: Observer):
        """Add an observer to the list"""
        self._observers.append(observer)
    
    def remove_observer(self, observer: Observer):
        """Remove an observer from the list"""
        if observer in self._observers:
            self._observers.remove(observer)
    
    def notify_observers(self):
        """Notify all observers of changes"""
        for observer in self._observers:
            observer.update(self._news)
    
    def set_news(self, news: str):
        """Set new news and notify observers"""
        self._news = news
        print(f"News Agency: Breaking news - {news}")
        self.notify_observers()
 
# Concrete Observer classes
class NewsChannel(Observer):
    def __init__(self, name: str):
        self.name = name
    
    def update(self, message: str):
        print(f"{self.name}: We just received news - {message}")
 
class Newspaper(Observer):
    def __init__(self, name: str):
        self.name = name
    
    def update(self, message: str):
        print(f"{self.name}: Publishing article about - {message}")
 
class MobileApp(Observer):
    def __init__(self, name: str):
        self.name = name
    
    def update(self, message: str):
        print(f"{self.name}: Push notification sent - {message}")
 
# Example usage
if __name__ == "__main__":
    # Create the subject (news agency)
    news_agency = NewsAgency()
    
    # Create observers
    cnn = NewsChannel("CNN")
    bbc = NewsChannel("BBC")
    times = Newspaper("The Times")
    news_app = MobileApp("NewsApp")
    
    # Register observers
    news_agency.add_observer(cnn)
    news_agency.add_observer(bbc)
    news_agency.add_observer(times)
    news_agency.add_observer(news_app)
    
    # Set news - all observers will be notified
    print("=== First News Update ===")
    news_agency.set_news("Major earthquake hits California")
    
    print("\n=== Second News Update ===")
    news_agency.set_news("New COVID variant discovered")
    
    # Remove an observer
    print("\n=== Removing BBC and updating again ===")
    news_agency.remove_observer(bbc)
    news_agency.set_news("Stock market reaches all-time high")