The 10 Minute Guide to the Observer Pattern in Python

Post to Twitter

Today I want to cover the Observer Pattern in Python. I’ll be using Python 3 but this should work equally well with Python 2.6.x and newer.

According to Wikipedia the Observer Pattern is: A software design pattern in which an object, called the subject, maintains a list of its dependents, called observers, and notifies them automatically of any state changes, usually by calling one of their methods. It is mainly used to implement distributed event handling systems.

Fortunately most of the code we will need can be found in the Wikipedia entry. I’ll modify this to suit my needs a little better.

First off create a new file called observable.py and add the following code:

class Observable(object):

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

    def register(self, observer):
        if not observer in self.observers:
            self.observers.append(observer)

    def unregister(self, observer):
        if observer in self.observers:
            self.observers.remove(observer)

    def unregister_all(self):
        if self.observers:
            del self.observers[:]

    def update_observers(self, *args, **kwargs):
        for observer in self.observers:
            observer.update(*args, **kwargs)

You can see I have a list of observers that I can add to, or remove from as needed. If a notification needs to go out then you can iterate through the observers and update each one. In this simple code every observer is notified, you could of course change the code easily to use a filter and only update particular observers that may only be interested in one or more particular kind of event/message.

Create the observer.py file now:

from abc import ABCMeta, abstractmethod

class Observer(object):
    __metaclass__ = ABCMeta

    @abstractmethod
    def update(self, *args, **kwargs):
        pass

Using Python’s Abstract Base Class this file and contents are essentially just an interface for classes to implement if your coming from a Java world. The point is we want to ensure those Observer based classes will contain the update method.

Let’s take a look at a couple classes now that makes use of all of this. In our fictional scenario we will assume we need to watch stock markets in America as well as Europe. When market conditions change we want all the markets to be notified. Obviously, each market is not in the same time zone so this example is a bit silly but let’s pretend for simplicity that European markets are online the same time as American markets and there are only two.

Create a file called app.py and add the following two classes and remaing code:

from observable import Observable
from observer import Observer


class AmericanStockMarket(Observer):
    def update(self, *args, **kwargs):
        print("American stock market received: {0}\n{1}".format(args, kwargs))


class EuropeanStockMarket(Observer):
    def update(self, *args, **kwargs):
        print("European stock market received: {0}\n{1}".format(args, kwargs))


if __name__ == "__main__":
    observable = Observable()

    american_observer = AmericanStockMarket()
    observable.register(american_observer)

    european_observer = EuropeanStockMarket()
    observable.register(european_observer)

    observable.update_observers('Market Rally', something='Hello World')

Run the app.py file and you should see the following:

American stock market received: ('Market Rally',)
{'something': 'Hello World'}
European stock market received: ('Market Rally',)
{'something': 'Hello World'}

You can see I dumped out both the args and kwargs but in reality you may use kwargs or something as a filter, or to deliver specific messages, etc. You can see how simple this pattern really is when you reduce it to its most basic foundations and that its a good way to update/notify other parts of your application of changes.

Let’s go ahead and generate some unit tests to ensure we have our methods covered. Create a file called observable_test.py and add the following code:

import unittest

from observable import Observable
from observer import Observer


class AnObserver(Observer):
    def __init__(self):
        self.args = []
        self.kwargs = {}

    def update(self, *args, **kwargs):
        self.args = args
        self.kwargs = kwargs
        return


class observerable_test(unittest.TestCase):

    def setUp(self):
        self.observable = Observable()

        self.observer1 = AnObserver()
        self.observer2 = AnObserver()
        self.observer3 = AnObserver()

    def test_remove_all(self):
        self.observable.register(self.observer1)
        self.observable.register(self.observer2)
        self.observable.register(self.observer3)

        self.observable.unregister_all()

        # Should be zero registered
        self.assertEqual(len(self.observable.observers), 0)

    def test_register(self):
        self.observable.register(self.observer1)
        self.observable.register(self.observer2)
        self.observable.register(self.observer3)

        # Should be three registered
        self.assertEqual(len(self.observable.observers), 3)
        self.observable.unregister_all()

    def test_unregister(self):
        self.observable.register(self.observer1)
        self.observable.register(self.observer2)
        self.observable.register(self.observer3)

        self.observable.unregister(self.observer2)
        self.observable.unregister(self.observer3)

        # Should be one registered
        self.assertEqual(len(self.observable.observers), 1)
        self.observable.unregister_all()

    def test_update_observers(self):
        self.observable.register(self.observer1)

        self.observable.update_observers('abc', msg='123')
        self.assertEqual(self.observer1.args[0], 'abc')
        self.assertEqual(self.observer1.kwargs['msg'], '123')
        self.observable.unregister_all()


if __name__ == "__main__":
    unittest.main()

This is a basic test and can be improved upon but it suits our needs to ensure everything is working.

Post to Twitter

This entry was posted in Design Patterns, Open Source, Python. Bookmark the permalink.

4 Responses to The 10 Minute Guide to the Observer Pattern in Python

  1. stu says:

    In python you may as well just use plain functions, the ‘Observer’ class isnt really needed.

    A lot of the GOF patterns disolve down untill there isnt much left in norr dynamic languages.

  2. moeabdol says:

    Great demonstration. This helped me a lot. Thank you.

  3. alex says:

    This is great example of what Java programmers are trying to do once they switch to Python – they feel like Python is missing all that crap and try to “port” it.

  4. alexo says:

    can you show example with gui and gui events?

Comments are closed.