Usage examples: The Observer pattern is pretty common in Python code, especially in the GUI components. It provides a way to react to events happening in other objects without coupling to their classes.
Identification: The pattern can be recognized by subscription methods, that store objects in a list and by calls to the update method issued to objects in that list.
Conceptual Example
This example illustrates the structure of the Observer design pattern. It focuses on answering these questions:
What classes does it consist of?
What roles do these classes play?
In what way the elements of the pattern are related?
main.py: Conceptual example
from__future__import annotationsfrom abc import ABC, abstractmethodfrom random import randrangefrom typing import ListclassSubject(ABC):""" The Subject interface declares a set of methods for managing subscribers. """@abstractmethoddefattach(self,observer: Observer) ->None:""" Attach an observer to the subject. """pass@abstractmethoddefdetach(self,observer: Observer) ->None:""" Detach an observer from the subject. """pass@abstractmethoddefnotify(self) ->None:""" Notify all observers about an event. """passclassConcreteSubject(Subject):""" The Subject owns some important state and notifies observers when the state changes. """ _state:int=None""" For the sake of simplicity, the Subject's state, essential to all subscribers, is stored in this variable. """ _observers: List[Observer]= []""" List of subscribers. In real life, the list of subscribers can be stored more comprehensively (categorized by event type, etc.). """defattach(self,observer: Observer) ->None:print("Subject: Attached an observer.") self._observers.append(observer)defdetach(self,observer: Observer) ->None: self._observers.remove(observer)""" The subscription management methods. """defnotify(self) ->None:""" Trigger an update in each subscriber. """print("Subject: Notifying observers...")for observer in self._observers: observer.update(self)defsome_business_logic(self) ->None:""" Usually, the subscription logic is only a fraction of what a Subject can really do. Subjects commonly hold some important business logic, that triggers a notification method whenever something important is about to happen (or after it). """print("\nSubject: I'm doing something important.") self._state =randrange(0, 10)print(f"Subject: My state has just changed to: {self._state}") self.notify()classObserver(ABC):""" The Observer interface declares the update method, used by subjects. """@abstractmethoddefupdate(self,subject: Subject) ->None:""" Receive update from subject. """pass"""Concrete Observers react to the updates issued by the Subject they had beenattached to."""classConcreteObserverA(Observer):defupdate(self,subject: Subject) ->None:if subject._state <3:print("ConcreteObserverA: Reacted to the event")classConcreteObserverB(Observer):defupdate(self,subject: Subject) ->None:if subject._state ==0or subject._state >=2:print("ConcreteObserverB: Reacted to the event")if__name__=="__main__":# The client code. subject =ConcreteSubject() observer_a =ConcreteObserverA() subject.attach(observer_a) observer_b =ConcreteObserverB() subject.attach(observer_b) subject.some_business_logic() subject.some_business_logic() subject.detach(observer_a) subject.some_business_logic()
Output.txt: Execution result
Subject: Attached an observer.
Subject: Attached an observer.
Subject: I'm doing something important.
Subject: My state has just changed to: 0
Subject: Notifying observers...
ConcreteObserverA: Reacted to the event
ConcreteObserverB: Reacted to the event
Subject: I'm doing something important.
Subject: My state has just changed to: 5
Subject: Notifying observers...
ConcreteObserverB: Reacted to the event
Subject: I'm doing something important.
Subject: My state has just changed to: 0
Subject: Notifying observers...
ConcreteObserverB: Reacted to the event