8.9. Observer
EN: Observer
PL: Obserwator
Type: object
The Observer design pattern is a behavioral design pattern that allows an object (known as the subject) to notify other objects (known as observers) about changes in its state. The subject maintains a list of observers and provides methods to add and remove observers from this list. When the state of the subject changes, it sends a notification to all its observers. This pattern is particularly useful in event-driven programming.
Here's a simple example of the Observer pattern in Python:
First, we define an abstract base class Observer that represents the
general interface for all observers:
>>> from abc import ABC, abstractmethod
>>>
>>> class Observer(ABC):
... @abstractmethod
... def update(self, message: str):
... pass
Then, we define a concrete observer class that implements the update
method:
>>> class ConcreteObserver(Observer):
... def update(self, message: str):
... print(f"ConcreteObserver: {message}")
Next, we define a Subject class that maintains a list of observers and
provides methods to add and remove observers. It also has a notify method
that sends a notification to all its observers:
>>> class Subject:
... def __init__(self):
... self.observers = []
...
... def register(self, observer: Observer):
... self.observers.append(observer)
...
... def unregister(self, observer: Observer):
... self.observers.remove(observer)
...
... def notify(self, message: str):
... for observer in self.observers:
... observer.update(message)
Finally, we can use the Subject and Observer classes like this:
>>> subject = Subject()
>>> observer = ConcreteObserver()
>>> subject.register(observer)
>>> subject.notify("Hello, Observer!")
ConcreteObserver: Hello, Observer!
In this example, ConcreteObserver is an observer that the Subject class
can notify. The Subject class doesn't need to know the details of how the
observers handle the notifications. It just calls the update method on
its observers.
8.9.1. Pattern
When the state of the object changes and you need to notify other objects about this change
Notify chart about changes in data to refresh
Spreadsheet formulas
Push or pull style of communication
8.9.2. Problem
8.9.3. Solution
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
class Observer(ABC):
@abstractmethod
def update(self) -> None:
pass
class Spreadsheet(Observer):
def update(self) -> None:
print('Spreadsheet got updated')
class Chart(Observer):
def update(self) -> None:
print('Chart got updated')
@dataclass
class Subject:
"""
Observable - class which is observed
"""
observers: list[Observer] = field(default_factory=list)
def add_observer(self, observer: Observer) -> None:
self.observers.append(observer)
def remove_observer(self, observer: Observer) -> None:
self.observers.remove(observer)
def notify_observers(self):
for observer in self.observers:
observer.update()
class DataSource(Subject):
value: int
def get_value(self) -> int:
return self.value
def set_value(self, value) -> None:
self.value = value
self.notify_observers()
if __name__ == '__main__':
datasource = DataSource()
sheet1 = Spreadsheet()
sheet2 = Spreadsheet()
chart = Chart()
datasource.add_observer(sheet1)
datasource.add_observer(sheet2)
datasource.add_observer(chart)
datasource.set_value(1)
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
class Observer(ABC):
@abstractmethod
def update(self) -> None:
pass
@dataclass
class Subject:
"""
Observable - class which is observed
"""
observers: list[Observer] = field(default_factory=list)
def add_observer(self, observer: Observer) -> None:
self.observers.append(observer)
def remove_observer(self, observer: Observer) -> None:
self.observers.remove(observer)
def notify_observers(self):
for observer in self.observers:
observer.update()
class DataSource(Subject):
value: int
def get_value(self) -> int:
return self.value
def set_value(self, value) -> None:
self.value = value
self.notify_observers()
@dataclass
class Spreadsheet(Observer):
datasource: DataSource
def update(self) -> None:
value = self.datasource.get_value()
print(f'Spreadsheet got updated: {value}')
@dataclass
class Chart(Observer):
datasource: DataSource
def update(self) -> None:
value = self.datasource.get_value()
print(f'Chart got updated: {value}')
if __name__ == '__main__':
datasource = DataSource()
sheet1 = Spreadsheet(datasource)
sheet2 = Spreadsheet(datasource)
chart = Chart(datasource)
datasource.add_observer(sheet1)
datasource.add_observer(sheet2)
datasource.add_observer(chart)
datasource.set_value(1)
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
class Observer(ABC):
@abstractmethod
def update(self, value: int) -> None:
pass
class Spreadsheet(Observer):
def update(self, value: int) -> None:
print(f'Spreadsheet got updated: {value}')
class Chart(Observer):
def update(self, value: int) -> None:
print(f'Chart got updated: {value}')
@dataclass
class Subject:
"""
Observable - class which is observed
"""
observers: list[Observer] = field(default_factory=list)
def add_observer(self, observer: Observer) -> None:
self.observers.append(observer)
def remove_observer(self, observer: Observer) -> None:
self.observers.remove(observer)
def notify_observers(self, value: int):
for observer in self.observers:
observer.update(value)
class DataSource(Subject):
value: int
def get_value(self) -> int:
return self.value
def set_value(self, value) -> None:
self.value = value
self.notify_observers(value)
if __name__ == '__main__':
datasource = DataSource()
sheet1 = Spreadsheet()
sheet2 = Spreadsheet()
chart = Chart()
datasource.add_observer(sheet1)
datasource.add_observer(sheet2)
datasource.add_observer(chart)
datasource.set_value(1)