Easy Python Patterns: The Observer pattern

Introduction

The observer pattern is a software design pattern that allows an object, usually called the subject, to maintain a list of dependents, called observers, and notify them automatically of any state-changes. Many languages have this pattern either built-in or in their standard libraries.

Examples:

C# has the concept of Observable in the standard .NET library. Also it has the concept of delegates.
Although Dart doesn’t have that, in Flutter we have the SetState method which automatically notifes any observer, and it is mainly used to refresh the UI.

It looks like this:

A breakdown of this diagram:

First there is the Observable, this could be an interface, the object which is to be observed. The observable holds a list of observers.
The Observer, which sends out a message to the observers. This could also be an interface.
The ConcreteObservable, the concrete class which holds the state. If anything changes in the state, the setState method is called, which in turn calls the notify method. That in turn notifies the observers.
The concrete observers which handle the state change.

Here the notify and update methods have no parameters, you could of course send extra data to the observers, and in most cases that will happen.

Implementation in Python

First we create an Observer class:

class Observer:
    def update(self):
        pass

The Observer class just has an update() method, which receives updates from the Observer‘s subject.

Talking of the subject, here is the implementation of the Subject class:

class Subject[T]:
    _observers: list[Observer] = None
    _state: T = None

    def __init__(self):
        self._observers = []

    def attach(self, new_observer: Observer):
        self._observers.append(new_observer)

    def detach(self, observer_to_remove: Observer):
        self._observers.remove(observer_to_remove)

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

    @property
    def state(self) -> T:
        return self._state

    @state.setter
    def state(self, new_state: T):
        self._state = new_state
        self.notify()

In the constructor we construct an empty list of observers, and a state
The attach() method attaches an observer to the subject, so the observer can receive updates
In the detach() the reverse happens
To send an update to the observers we use the notify() method. In it we iterate over the list of observers, and send each an update() message
We use the @property attribute for setting and getting the state. Note that in the setter we call the notify() method to notify the observers of a state change. The beauty of this is, is that this is completely transparent to the clients.

Next we implement a couple of observers:

class ConcreteObserver1(Observer):
    _subject: Subject = None

    def __init__(self, new_subject: Subject):
        self._subject = new_subject

    def update(self):
        print(f"ConcreteObserver1 received notification: {self._subject.state}")


class ConcreteObserver2(Observer):
    _subject: Subject = None

    def __init__(self, new_subject: Subject):
        self._subject = new_subject

    def update(self):
        print(f"ConcreteObserver2 received notification: {self._subject.state}")

Some notes:

Each observer has a subject from which it receives update. In the constructor we set the subject
In the update we print out the state of the observed subject.

Time to test

We can now test our simple setup:

if __name__ == "__main__":
    subject: Subject = Subject()
    observer1: ConcreteObserver1 = ConcreteObserver1(subject)
    observer2: ConcreteObserver2 = ConcreteObserver2(subject)
    subject.attach(observer1)
    subject.attach(observer2)
    subject.state = "state 1"
    subject.state = "state 2"

    subject.detach(observer2)

    subject.state = "state 3"

You should output like this:

ConcreteObserver1 received notification: state 1
ConcreteObserver2 received notification: state 1
ConcreteObserver1 received notification: state 2
ConcreteObserver2 received notification: state 2
ConcreteObserver1 received notification: state 3

Conclusion

As you can see, the Observer pattern is not very difficult to implement, yet very powerful. It can be used to communicate safely between disparate portions of your program.