Observer in Python
Observer is a behavioral design pattern that allows some objects to notify other objects about changes in their state.
The Observer pattern provides a way to subscribe and unsubscribe to and from these events for any object that implements a subscriber interface.
Complexity:
Popularity:
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 annotations
from abc import ABC, abstractmethod
from random import randrange
from typing import List
class Subject(ABC):
"""
The Subject interface declares a set of methods for managing subscribers.
"""
@abstractmethod
def attach(self, observer: Observer) -> None:
"""
Attach an observer to the subject.
"""
pass
@abstractmethod
def detach(self, observer: Observer) -> None:
"""
Detach an observer from the subject.
"""
pass
@abstractmethod
def notify(self) -> None:
"""
Notify all observers about an event.
"""
pass
class ConcreteSubject(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.).
"""
def attach(self, observer: Observer) -> None:
print("Subject: Attached an observer.")
self._observers.append(observer)
def detach(self, observer: Observer) -> None:
self._observers.remove(observer)
"""
The subscription management methods.
"""
def notify(self) -> None:
"""
Trigger an update in each subscriber.
"""
print("Subject: Notifying observers...")
for observer in self._observers:
observer.update(self)
def some_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()
class Observer(ABC):
"""
The Observer interface declares the update method, used by subjects.
"""
@abstractmethod
def update(self, subject: Subject) -> None:
"""
Receive update from subject.
"""
pass
"""
Concrete Observers react to the updates issued by the Subject they had been
attached to.
"""
class ConcreteObserverA(Observer):
def update(self, subject: Subject) -> None:
if subject._state < 3:
print("ConcreteObserverA: Reacted to the event")
class ConcreteObserverB(Observer):
def update(self, subject: Subject) -> None:
if subject._state == 0 or 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
