The Observer Pattern in Python
I find that the best way to get a deep understanding of a programming concept is to code it up yourself. Here's my take on the standard Observer Pattern. I wanted to fully understand it, because I expect to need it shortly. This also gave me the opportunity to learn about the handy-dandy weakref module.
This implementation is a bit different from the standard GoF one. For one thing, I wanted a clean way for an Observer to be able to observe multiple Observables, while being able to easily know which Observable was sending it an update notification. That's why, when registering an Observer with an Observable, the Observer tells the Observable the name of the Observer's function to call for the update notification.
Another difference has to do with the methods that the Observer base class provides. Now, strictly speaking from a Python perspective, the Observer base class is unnecessary. Any class that provides a method with the right signature can be used as an Observer. But when I was talking out this design in my head, I kept using phrases like "The observer tells the observable that it is interested in it." This really seemed like an action performed by the Observer, so I wanted that behaviour in an Observer base class.
The last big difference has to do with the way an Observer subscribes and unsubscribes to update notification by an Observable. The standard GoF implementation has the Observable class having 'attach' and 'detach' methods. I have an 'attach' method (named addObserver), and a 'detach' method (named removeObserver). But I'm also using weakref.WeakKeyDictionary as the way that an Observable keeps a list of its Observers. This means that the fact that there's a reference to the Observer in the Observable does not keep the Observable alive if it goes out of scope or is deleted. The weakref to the Observer will be automatically deleted from the Observable's list (I know, a dictionary is not a list. I'm using the word 'list' here loosely.) So, the 'removeObserver' method is not strictly needed; just delete an Observer, and it gets removed from the list kept by any Observable it subscribed to.
Here's the code. It should go in a file named observer.py: import weakref
import types ##
# The Observer Pattern in Python
#
# Design goals:
# 1. An Observer should be able to observe multiple Observables.
# 2. An Observer should be able to tell an Observable what kinds of
# events it is interested in observing.
# 3. When an Observer is deleted, all Observables that it is observing
# should be notified to remove that Observer.
# 4. When an Observer is notified of an event,
# it should be able to tell which Observable is calling it,
# and which event occured.
## ##
# The abstract Observable Class.
# class Observable(object):
def __init__(self):
# A WeakKeyDictionary is one where, if the object used as the key
# gets deleted, automatically removes that key from the
# dictionary. Thus, any Observers which get deleted will be
# automatically removed from the observers dictionary, thus having
# two effects:
# We won't have references to zombie objects in the dictionary, and
# We won't have zombie objects, because the reference in this
# dictionary won't stay around, and so won't keep the deleted object
# alive.
self._observers = weakref.WeakKeyDictionary() ##
# Add an observer to this Observable's notification list.
# @param observer The Observer to add.
# @param cbname The name (as a string) of the Observer's
# method to call for an event notification, or None for the default
# "update" method.
# @param events The events the Observer is interested in being
# notified about. None means all events. def addObserver(self, observer, cbname=None, events=None):
if cbname is None:
cbname = "update"
if events is not None and type(events) not in (types.TupleType,
types.ListType):
events = (events,)
self._observers[observer] = (cbname, events) ##
# Remove an observer from this Observable's list of observers.
# Note that this function is not strictly required, because when a
# registered Observer is deleted, the weakref mechanism will cause
# it to be removed from the notification list.
# @param observer the Observer to remove. def removeObserver(self, observer):
if observer in self._observers:
del self._observers[observer] ##
# Notify all currently-registered Observers.
# Each observer must have an 'update' method, which should take
# three parameters (in addition to self): the Observable, an event,
# and a message.
# This method will be
# called if the event is one that the Observer is interested in,
# or if event is 'None', or if the Observer is interested in all
# events (it was registered with an event list of 'None').
# @param event The event to notify the Observers about. None
# means no specific event.
# @param msg A reference to any data object that should be passed
# to the Observers, or None. def notifyObservers(self, event=None, msg=None):
for observer, data in self._observers.items():
#print "data is", data
cbname, events = data
#print "cbname is", cbname
#print "events is", events
if events is None or event is None or event in events:
cb = getattr(observer, cbname, None)
if cb is None:
raise NotImplementedError, "Observer has no %s method." %
cbname
cb(self, event, msg)
##
# The abstract Observer Class
# This class is a mix-in to add Observable registration methods to
# a concrete Observer class. It is not strictly required.
#
class Observer(object): ##
# @param observable The Observable to observe.
# @param The events this Observer is interested in being
# notified about. This should be a tuple or list of events. def __init__(self, observable=None, cbname=None, events=None):
if (observable is not None):
observable.addObserver(self, cbname, events) ##
# Inform an Observable that you would like to be notified when
# an interesting event occurs in the Observable.
# @param observable The Observable this Observer would like
# to observe.
# @param cbname The name (as a string) of the Observer's
# method to call for an event notification, or None for the default
# "update" method.
# @param events A tuple or list of events this Observer would like
# to be notified of by the Observer, or None if it would like to
# be notified of all events. def subscribeToObservable(self, observable, cbname=None, events=None):
assert observable is not None, "Observable is None"
observable.addObserver(self, cbname, events) ##
# Inform an observable that this Observer is no longer interested in it.
# Note that this function is not strictly required, because when a
# registered Observer is deleted, the weakref mechanism will cause
# it to be removed from the Observable's notification list.
# Use this function when you want to unsubscribe an Observer
# without deleting it.
# @param observable The Observable that this Observer no longer wants
# to observe. def unsubscribeToObservable(self, observable):
assert observable is not None, "Observable is None"
observable.removeObserver(self)
Here's the code for the unit tests, which also serves as usage examples. It should go in a file named test_observer.py: #! /usr/bin/env python import unittest
from observer import Observable, Observer EVENT_FOO, EVENT_UPDATE = range(2) class Stuff(Observable): def __init__(self):
Observable.__init__(self)
self._data = None def setData(self, data):
self._data = data
self.notifyObservers(None, data) def setDataWithUpdateEvent(self, data):
self._data = data
self.notifyObservers(EVENT_UPDATE, data) def setDataWithFooEvent(self, data):
self._data = data
self.notifyObservers(EVENT_FOO, data)
class StuffWatcher(Observer):
def __init__(self, observable=None, cbname=None, events=None):
Observer.__init__(self, observable, cbname, events)
self._data = None
self._updateData = None
self._reportData = None def update(self, observable, event, msg):
self._data = msg
self._updateData = msg def report(self, observable, event, msg):
self._data = msg
self._reportData = msg
class TestCase_01_Observer(unittest.TestCase):
def test_01_noMethod(self):
observable = Stuff()
observer1 = StuffWatcher(observable, "foo")
self.assertRaises(NotImplementedError, observable.setData, 10) def test_02_simple(self): observable = Stuff()
observer1 = StuffWatcher(observable)
observer2 = StuffWatcher(observable)
self.assertEqual(len(observable._observers), 2) observable.setData(10)
self.assertEqual(observer1._data, 10)
self.assertEqual(observer2._data, 10) observable.setData(20)
self.assertEqual(observer1._data, 20)
self.assertEqual(observer2._data, 20) del observer1
self.assertEqual(len(observable._observers), 1) observable.setData(30)
self.assertEqual(observer2._data, 30) observer3 = StuffWatcher()
self.assertEqual(len(observable._observers), 1)
observer3.subscribeToObservable(observable)
self.assertEqual(len(observable._observers), 2)
observable.setData(40)
self.assertEqual(observer2._data, 40)
self.assertEqual(observer3._data, 40) del observer2
self.assertEqual(len(observable._observers), 1) observable.setData(50)
self.assertEqual(observer3._data, 50) def test_03_specificEvents(self): observable = Stuff()
observer1 = StuffWatcher(observable, events=EVENT_UPDATE)
observer2 = StuffWatcher(observable, events=EVENT_FOO) observable.setDataWithUpdateEvent(10)
self.assertEqual(observer1._data, 10)
self.assertEqual(observer2._data, None) observable = Stuff()
observer1 = StuffWatcher(observable, events=(EVENT_UPDATE, EVENT_FOO)) observable.setDataWithUpdateEvent(10)
self.assertEqual(observer1._data, 10) observable.setDataWithFooEvent(20)
self.assertEqual(observer1._data, 20) def test_03_multipleObservables(self):
observable1 = Stuff()
observable2 = Stuff()
observer = StuffWatcher() observer.subscribeToObservable(observable1, "update", EVENT_UPDATE)
observer.subscribeToObservable(observable2, "report", EVENT_FOO)
self.assertEqual(len(observable1._observers), 1)
self.assertEqual(len(observable2._observers), 1) observable1.setDataWithUpdateEvent(10)
self.assertEqual(observer._data, 10)
self.assertEqual(observer._updateData, 10)
self.assertEqual(observer._reportData, None) observable1.setDataWithFooEvent(20)
self.assertEqual(observer._data, 10)
self.assertEqual(observer._updateData, 10)
self.assertEqual(observer._reportData, None) observable2.setDataWithUpdateEvent(30)
self.assertEqual(observer._data, 10)
self.assertEqual(observer._updateData, 10)
self.assertEqual(observer._reportData, None) observable2.setDataWithFooEvent(40)
self.assertEqual(observer._data, 40)
self.assertEqual(observer._updateData, 10)
self.assertEqual(observer._reportData, 40)
if __name__ == "__main__":
unittest.main()
4:11:59 PM 
|
