Carlo said:
hello there
I'd like to take control over the method dispatching of every subclass
of a given python class. Currently I've redefined __getattribute__()
so i can find the right instance/method to dispatch to. This works
fine, albeit it may seem a bit hackish. Further I want to control the
actual dispatching, so that I can execute methods before, after or
instead of the actual method. I would prefer to do this with the least
modification of the source due to debug'ing issues. My best idea so
far is to rename every method who subclass the magic class X and then
create new methods in place of the renamed ones, which looks up in a
list to execute "before-methods", then execute the renamed method etc..
Only I don't know how to do this. I've found some MOP code which was
designed for python old-style objects... but this is not what I want.
Can anyone kindly give me some pointers for help or some actual code?
many thanks in advance..
You don't need metaclasses at all, actually. I attached the action
module that I created at work. You can set callbacks in any class:
.... def foo(self, arg):
.... print 'C.foo(%s)' % arg
........ print 'callback(%s)' % arg
....C.foo(10)
callback(10)
Besides after(), you have also before(). Also there's result(), where
the callback receives as parameter the return of the given function.
HTH,
Nicodemus.
'''
Implements non-intrusive callbacks for methods.
Note: this module doesn't work with Qt objects, and with methods that are
attached to a property (setting a property won't trigger the callback!)
'''
#==============================================================================
# before
#==============================================================================
def before(method, *callbacks):
'''Registers the given callbacks to be execute before the given method is
called, with the same arguments. The method can be eiher an unbound method
or a bound method. If it is an unbound method, *all* instances of the class
will generate callbacks when method is called. If it is a bound method, only
the method of the instance will generate callbacks.
'''
wrapper = _Setup(method)
for callback in callbacks:
if callback not in wrapper.before:
wrapper.before.append(callback)
#==============================================================================
# after
#==============================================================================
def after(method, *callbacks):
'''Same as before, but will callback after method is executed.
'''
wrapper = _Setup(method)
for callback in callbacks:
if callback not in wrapper.after:
wrapper.after.append(callback)
#==============================================================================
# result
#==============================================================================
def result(method, *callbacks):
'''The callbacks are called after the given method is executed, and the
callbacks will receive the return value of the method as parameter.
'''
wrapper = _Setup(method)
for callback in callbacks:
if callback not in wrapper.result:
wrapper.result.append(callback)
#==============================================================================
# remove
#==============================================================================
def remove(method, callback):
'''Removes the given callback from a method previously connected using
after or before.
'''
if hasattr(method, 'before') and hasattr(method, 'after'):
try:
method.before.remove(callback)
except ValueError: pass
try:
method.after.remove(callback)
except ValueError: pass
try:
method.result.remove(callback)
except ValueError: pass
#==============================================================================
# Internals
#==============================================================================
def _MakeWrapper(method):
'''Creates a function with two lists of callbacks, before and after.
This function when called, will call all the "before" callbacks, call
the original method, and then call all the "after" callbacks.
'''
def wrapper(*args, **kwargs):
# call all the before callbacks
for callback in wrapper.before:
callback(*args, **kwargs)
# call original method
result = method(*args, **kwargs)
# call all the after callbacks
for callback in wrapper.after:
callback(*args, **kwargs)
# call all result callbacks
for callback in wrapper.result:
callback(result)
return result
wrapper.before = []
wrapper.after = []
wrapper.result = []
return wrapper
def _Setup(method):
'''Generates a wrapper for the given method, or returns the method itself
if it is already a wrapper.
'''
if hasattr(method, 'before') and hasattr(method, 'after'):
# its a wrapper already
return method
else:
wrapper = _MakeWrapper(method)
if method.im_self is None:
# override the class method
target = method.im_class
else:
# override the instance method
target = method.im_self
setattr(target, method.__name__, wrapper)
return wrapper
#==============================================================================
# property
#==============================================================================
class property(object):
'''Creates an property just like a built-in property, except that it can
be used with the actions in this module.
'''
def __init__(self, fget=None, fset=None, fdel=None, doc=None):
self.fget_name = fget and fget.__name__
self.fset_name = fset and fset.__name__
self.fdel_name = fdel and fdet.__name__
self.doc = doc or ''
def FromNames(cls, fget_name, fset_name=None, fdel_name=None, doc=None):
self = cls()
self.fget_name = fget_name
self.fset_name = fset_name
self.fdel_name = fdel_name
self.doc = doc or ''
return self
FromNames = classmethod(FromNames)
def __get__(self, obj, objtype=None):
if obj is None:
return self
if self.fget_name is None:
raise AttributeError, "unreadable attribute"
fget = getattr(obj, self.fget_name)
return fget()
def __set__(self, obj, value):
if self.fset_name is None:
raise AttributeError, "can't set attribute"
fset = getattr(obj, self.fset_name)
fset(value)
def __delete__(self, obj):
if self.fdel_name is None:
raise AttributeError, "can't delete attribute"
fdel = getattr(obj, self.fdel_name)
fdel()
#==============================================================================
# Actions
#==============================================================================
class Actions(object):
'''Holds the connections created, making easy to disconnect later.
'''
def __init__(self):
self._actions = []
def before(self, sender, *callbacks):
before(sender, *callbacks)
for callback in callbacks:
self._actions.append((sender, callback))
def after(self, sender, *callbacks):
after(sender, *callbacks)
for callback in callbacks:
self._actions.append((sender, callback))
def result(self, sender, *callbacks):
result(sender, *callbacks)
for callback in callbacks:
self._actions.append((sender, callback))
def remove(self, sender, callback):
remove(sender, callback)
for s, c in self._actions:
if sender == s and callback == c:
self._actions.remove((sender, callback))
break
def remove_all(self):
for sender, callback in self._actions:
remove(sender, callback)
self._actions[:] = []
def remove_sender(self, sender):
for index in xrange(len(self._actions)-1, -1, -1):
s, c = self._actions[index]
if s == sender:
remove(s, c)
self._actions.pop(index)
def remove_callback(self, callback):
for index in xrange(len(self._actions)-1, -1, -1):
s, c = self._actions[index]
if c == callback:
remove(s, c)
self._actions.pop(index)
#==============================================================================
# tests
#==============================================================================
if __name__ == '__main__':
import unittest
class ActionTest(unittest.TestCase):
def setUp(self):
class C:
def foo(s, arg):
self.foo_called = (s, arg)
return arg
self.C = C
self.a = C()
self.b = C()
self.after_count = 0
self.before_count = 0
def after(*args):
self.after_called = args
self.after_count += 1
self.after = after
def before(*args):
self.before_called = args
self.before_count += 1
self.before = before
def result(arg):
self.result_arg = arg
self.result = result
def testClassOverride(self):
before(self.C.foo, self.before)
after(self.C.foo, self.after)
result(self.C.foo, self.result)
self.a.foo(1)
self.assertEqual(self.foo_called, (self.a, 1))
self.assertEqual(self.after_called, (self.a, 1))
self.assertEqual(self.after_count, 1)
self.assertEqual(self.before_called, (self.a, 1))
self.assertEqual(self.before_count, 1)
self.assertEqual(self.result_arg, 1)
self.b.foo(2)
self.assertEqual(self.foo_called, (self.b, 2))
self.assertEqual(self.after_called, (self.b, 2))
self.assertEqual(self.after_count, 2)
self.assertEqual(self.before_called, (self.b, 2))
self.assertEqual(self.before_count, 2)
self.assertEqual(self.result_arg, 2)
remove(self.C.foo, self.before)
self.a.foo(3)
self.assertEqual(self.foo_called, (self.a, 3))
self.assertEqual(self.after_called, (self.a, 3))
self.assertEqual(self.after_count, 3)
self.assertEqual(self.before_called, (self.b, 2))
self.assertEqual(self.before_count, 2)
self.assertEqual(self.result_arg, 3)
def testInstanceOverride(self):
before(self.a.foo, self.before)
after(self.a.foo, self.after)
result(self.a.foo, self.result)
self.a.foo(1)
self.assertEqual(self.foo_called, (self.a, 1))
self.assertEqual(self.after_called, (1,))
self.assertEqual(self.before_called, (1,))
self.assertEqual(self.after_count, 1)
self.assertEqual(self.before_count, 1)
self.assertEqual(self.result_arg, 1)
self.b.foo(2)
self.assertEqual(self.foo_called, (self.b, 2))
self.assertEqual(self.after_called, (1,))
self.assertEqual(self.before_called, (1,))
self.assertEqual(self.after_count, 1)
self.assertEqual(self.before_count, 1)
self.assertEqual(self.result_arg, 1)
remove(self.a.foo, self.before)
self.a.foo(2)
self.assertEqual(self.foo_called, (self.a, 2))
self.assertEqual(self.after_called, (2,))
self.assertEqual(self.before_called, (1,))
self.assertEqual(self.after_count, 2)
self.assertEqual(self.before_count, 1)
self.assertEqual(self.result_arg, 2)
before(self.a.foo, self.before, self.before)
remove(self.a.foo, self.result)
self.a.foo(5)
self.assertEqual(self.before_called, (5,))
self.assertEqual(self.before_count, 2)
self.assertEqual(self.result_arg, 2)
def testActionProperty(self):
class C(object):
def setx(self, x): self._x = x
def getx(self): return self._x
x = property(getx, setx)
def beforex(x):
self.beforex_called = x
def afterx(x):
self.afterx_called = x
c = C()
before(c.setx, beforex)
after(c.setx, afterx)
c.x = 2
self.assertEqual(c.x, 2)
self.assertEqual(self.afterx_called, 2)
self.assertEqual(self.beforex_called, 2)
class C(object):
def setx(self, x): self._x = x
def getx(self): return self._x
x = property.FromNames('getx', 'setx')
c = C()
before(c.setx, beforex)
after(c.setx, afterx)
c.x = 2
self.assertEqual(c.x, 2)
self.assertEqual(self.afterx_called, 2)
self.assertEqual(self.beforex_called, 2)
unittest.main()
