Observer Pattern

Issue

• One or more objects needs to know. when a different objects (Subject) changes its state

• When all these objects need to ask the Subject object individually for potential changes (e.g. every 10 ms), there will be a lot of queries (Polling) -> lots of queries

Solution

• The Subject pushes change information to its Observers (subscribers) objects

• all Observer objects need to register for a Subject object to receive changes

• when a change occurs in Subject the information is pushed to all registered Oberserver objects

Note

Subject class will be referred to as Observable from here on for clearer naming.

Definition

The Observer Pattern defines a one-to-many dependency between object so that when one object changes state, all of its dependencies are notified and updated automatically.

        classDiagram
    Observer "0..*" <-- Observable : HAS-A
    Observer <|-- ConcreteObserver : implements
    Observable <|-- ConcreteObservable : implements
    ConcreteObservable <-- ConcreteObserver : HAS-A
    class Observable {
        <<interface>>
        +add(Observer)
        +remove(Observer)
        +notify()
    }
    class Observer {
        <<interface>>
        +update()
    }
    class ConcreteObservable {
        +add(Observer)
        +remove(Observer)
        +notify()
        +get_state()
    }
    class ConcreteObserver {
        +update()
    }
    

• a ConcreteObservable updates its registered ConcreteObservers by calling their update() method

• a ConcreteObservable class has additional methods to do create or obtain the data, it is supposed to push towards its registered ConcreteObservers

Warning

The ConcreteObservable class violates the Open-Closed Principle since it handles the notification of its registered ConcreteObservers and has other logic of whatever it is the object is doing (e.g. pulling some data from a weather station), here bound to the method get_state().

• a ConcreteObservable passes itself into the constructor of ConcreteObserver. This allows the ConcreteObservable to call the update() method without passing itself or the updated data, since the ConcreteObserver can reference it itself

Example

Find a template example for Python at https://refactoring.guru/design-patterns/observer/python/example.

A weather station, which contains a sensor to measure the temperature (and other data) and multiple devices, which each display the data.

        classDiagram
    Observer "0..*" <-- Observable : HAS-A
    Observer <|-- PhoneDisplay : implements
    Observer <|-- DeskDisplay : implements
    Observable <|-- WeatherStation : implements
    WeatherStation <-- PhoneDisplay : HAS-A
    WeatherStation <-- DeskDisplay : HAS-A
    class Observable {
        <<interface>>
        +add(Observer)
        +remove(Observer)
        +notify()
    }
    class Observer {
        <<interface>>
        +update()
    }
    class WeatherStation {
        +add(Observer)
        +remove(Observer)
        +notify()
    }
    class PhoneDisplay {
        +update()
    }
    class DeskDisplay {
        +update()
    }
    

from abc import ABC, abstractmethod


class Observer(ABC):

    @abstractmethod
    def update(self):
        pass


class Observable(ABC):

    @abstractmethod
    def add(self, observer: Observer):
        pass

    @abstractmethod
    def remove(self, observer: Observer):
        pass

    @abstractmethod
    def notify(self):
        pass


class WeatherStation(Observable):

    observers = []
    temperature = 20

    def add(self, observer: Observer):
        self.observers.append(observer)

    def remove(self, observer: Observer):
        self.observers.remove(observer)

    def notify(self):
        for observer in self.observers:
            observer.update()

    def get_temperature(self):
        return self.temperature

    # add method to retrieve temperature from sensor


class PhoneDisplay(Observer):

    temperature = None

    def __init__(self, weather_station: WeatherStation):
        self.station = weather_station

    def update(self):
        self.temperature = self.station.get_temperature()

    def display(self):
        print(f"Phone Display: {self.temperature}")

    # some additional method(s) on how to display temperature


class DeskDisplay(Observer):

    temperature = None

    def __init__(self, weather_station: WeatherStation):
        self.station = weather_station

    def update(self):
        self.temperature = self.station.get_temperature()

    def display(self):
        print(f"Desk Display: {self.temperature}")

    # some additional method(s) on how to display temperature


if __name__ == "__main__":

    station = WeatherStation()
    phone_display = PhoneDisplay(station)
    desk_display = DeskDisplay(station)

    station.add(phone_display)
    station.add(desk_display)

    # initial display
    phone_display.update()
    desk_display.update()
    phone_display.display()
    desk_display.display()

    # mocking a temperature change
    station.temperature = 25
    station.notify()

    # updated display
    phone_display.display()
    desk_display.display()