I have not yet seen an implementation of the observer pattern in Python which satisfies the following criteria:
- A thing which is observed should not keep its observers alive if all other references to those observers disappear.
- Adding and removing observers should be pythonic. For example, if I have an object
foo
with a bound method.bar
, I should be able to add an observer by calling a method on the method, like this:foo.bar.addObserver(observer)
. - I should be able to make any method observable without subclassing. For example, I should be able to make a method observable just by decorating it.
- This must work for types which are unhashable, and must be able to be used on an arbitrary number of methods per class.
- The implementation should be comprehensible to other developers.
Here is my attempt (now on github), which allows bound methods to observe other bound methods (see the test example below of example usage):
import weakref
import functools
class ObservableMethod(object):
"""
A proxy for a bound method which can be observed.
I behave like a bound method, but other bound methods can subscribe to be
called whenever I am called.
"""
def __init__(self, obj, func):
self.func = func
functools.update_wrapper(self, func)
self.objectWeakRef = weakref.ref(obj)
self.callbacks = {} #observing object ID -> weak ref, methodNames
def addObserver(self, boundMethod):
"""
Register a bound method to observe this ObservableMethod.
The observing method will be called whenever this ObservableMethod is
called, and with the same arguments and keyword arguments. If a
boundMethod has already been registered to as a callback, trying to add
it again does nothing. In other words, there is no way to sign up an
observer to be called back multiple times.
"""
obj = boundMethod.__self__
ID = id(obj)
if ID in self.callbacks:
s = self.callbacks[ID][1]
else:
wr = weakref.ref(obj, Cleanup(ID, self.callbacks))
s = set()
self.callbacks[ID] = (wr, s)
s.add(boundMethod.__name__)
def discardObserver(self, boundMethod):
"""
Un-register a bound method.
"""
obj = boundMethod.__self__
if id(obj) in self.callbacks:
self.callbacks[id(obj)][1].discard(boundMethod.__name__)
def __call__(self, *arg, **kw):
"""
Invoke the method which I proxy, and all of it's callbacks.
The callbacks are called with the same *args and **kw as the main
method.
"""
result = self.func(self.objectWeakRef(), *arg, **kw)
for ID in self.callbacks:
wr, methodNames = self.callbacks[ID]
obj = wr()
for methodName in methodNames:
getattr(obj, methodName)(*arg, **kw)
return result
@property
def __self__(self):
"""
Get a strong reference to the object owning this ObservableMethod
This is needed so that ObservableMethod instances can observe other
ObservableMethod instances.
"""
return self.objectWeakRef()
class ObservableMethodDescriptor(object):
def __init__(self, func):
"""
To each instance of the class using this descriptor, I associate an
ObservableMethod.
"""
self.instances = {} # Instance id -> (weak ref, Observablemethod)
self._func = func
def __get__(self, inst, cls):
if inst is None:
return self
ID = id(inst)
if ID in self.instances:
wr, om = self.instances[ID]
if not wr():
msg = "Object id %d should have been cleaned up"%(ID,)
raise RuntimeError(msg)
else:
wr = weakref.ref(inst, Cleanup(ID, self.instances))
om = ObservableMethod(inst, self._func)
self.instances[ID] = (wr, om)
return om
def __set__(self, inst, val):
raise RuntimeError("Assigning to ObservableMethod not supported")
def event(func):
return ObservableMethodDescriptor(func)
class Cleanup(object):
"""
I manage remove elements from a dict whenever I'm called.
Use me as a weakref.ref callback to remove an object's id from a dict
when that object is garbage collected.
"""
def __init__(self, key, d):
self.key = key
self.d = d
def __call__(self, wr):
del self.d[self.key]
Here is a test routine, which also serves to illustrate use of the code:
def test():
buf = []
class Foo(object):
def __init__(self, name):
self.name = name
@event
def bar(self):
buf.append("%sbar"%(self.name,))
def baz(self):
buf.append("%sbaz"%(self.name,))
a = Foo('a')
assert len(Foo.bar.instances) == 0
# Calling an observed method adds the calling instance to the descriptor's
# instances dict.
a.bar()
assert buf == ['abar']
buf = []
assert len(Foo.bar.instances) == 1
assert Foo.bar.instances.keys() == [id(a)]
assert len(a.bar.callbacks) == 0
b = Foo('b')
assert len(Foo.bar.instances) == 1
b.bar()
assert buf == ['bbar']
buf = []
assert len(Foo.bar.instances) == 2
# Methods added as observers are called when the observed method runs
a.bar.addObserver(b.baz)
assert len(a.bar.callbacks) == 1
assert id(b) in a.bar.callbacks
a.bar()
assert buf == ['abar','bbaz']
buf = []
# Observable methods can sign up as observers
mn = a.bar.callbacks[id(b)][1]
a.bar.addObserver(b.bar)
assert len(a.bar.callbacks) == 1
assert len(mn) == 2
assert 'bar' in mn
assert 'baz' in mn
a.bar()
buf.sort()
assert buf == ['abar','bbar','bbaz']
buf = []
# When an object is destroyed it is unregistered from any methods it was
# observing, and is removed from the descriptor's instances dict.
del b
assert len(a.bar.callbacks) == 0
a.bar()
assert buf == ['abar']
buf = []
assert len(Foo.bar.instances) == 1
del a
assert len(Foo.bar.instances) == 0