YouTip

The Observer pattern is a behavioral design pattern that defines a one-to-many dependency relationship, allowing multiple observer objects to simultaneously listen to a subject object. When the subject object's state changes, all its observers are notified and updated automatically. ### Observer Pattern in Real Life Imagine a newspaper subscription scenario: * **Newspaper office** (Subject) publishes news * **Subscribers** (Observers) subscribe to the newspaper * When a new newspaper is published, all subscribers automatically receive it * Subscribers can cancel their subscription at any time This "publish-subscribe" mechanism is a typical application of the Observer pattern. * * * ## Core Components of the Observer Pattern The Observer pattern contains four key roles: ### 1. Subject * Maintains a list of observers * Provides methods to register and unregister observers * Notifies all observers of state changes ### 2. Concrete Subject * Stores specific state information * Notifies observers when the state changes ### 3. Observer * Defines an update interface for receiving subject notifications ### 4. Concrete Observer * Implements the update interface to keep its state consistent with the subject's state * * * ## Implementation of the Observer Pattern Let's understand the implementation of the Observer pattern through a complete example. ### Basic Implementation ## Example from abc import ABC, abstractmethod from typing import List # Observer interface class Observer(ABC): @abstractmethod def update(self, message: str): """Observer update method""" pass # Subject interface class Subject(ABC): @abstractmethod def register_observer(self, observer: Observer): """Register observer""" pass @abstractmethod def remove_observer(self, observer: Observer): """Remove observer""" pass @abstractmethod def notify_observers(self): """Notify all observers""" pass # Concrete Subject - News Publisher class NewsPublisher(Subject): def __init__ (self): self._observers: List=[] self._latest_news ="" def register_observer(self, observer: Observer): """Register observer""" if observer not in self._observers: self._observers.append(observer) print(f"Observer {observer.__class__.__name__} registered successfully") def remove_observer(self, observer: Observer): """Remove observer""" if observer in self._observers: self._observers.remove(observer) print(f"Observer {observer.__class__.__name__} removed successfully") def notify_observers(self): """Notify all observers""" print(f"\n Publishing new message: {self._latest_news}") for observer in self._observers: observer.update(self._latest_news) def publish_news(self, news: str): """Publish news""" self._latest_news = news self.notify_observers() def get_observer_count(self) ->int: """Get observer count""" return len(self._observers) # Concrete Observer - Email Subscriber class EmailSubscriber(Observer): def __init__ (self, name: str): self.name= name def update(self, message: str): """Receive update and send email""" print(f" {self.name} received message: {message}") # Concrete Observer - SMS Subscriber class SMSSubscriber(Observer): def __init__ (self, name: str, phone: str): self.name= name