9.8. 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.

9.8.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

../../_images/designpatterns-observer-pattern.png

../../_images/designpatterns-observer-push.png
../../_images/designpatterns-observer-pull.png

9.8.2. Problem

design-patterns/behavioral/img/designpatterns-observer-problem.png


9.8.3. Solution

../../_images/designpatterns-observer-solution.png

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)
Code 9.1. push
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)
Code 9.2. pull
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)

9.8.4. Assignments