There are a couple of other AoP related recipes on ASPN. This one is designed to have a simple, elegant interface.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 | import functools
import sys
import types
def advise(*join_points):
"""Hook advice to a function or method.
advise() is a decorator that takes a set of functions or methods and
injects the decorated function in their place. The decorated function should
have the signature:
@advise(some_function, Class.some_method, Class.some_class_method,
Class.some_static_method, ...)
def interceptor(on, next, *args, **kwargs):
...
Where "on" is the object that hosts the intercepted function (ie. a module,
class or instance) and "next" is the next function in the interception
chain.
>>> def eat(lunch):
... print 'eating', lunch
>>> @advise(eat)
... def replace_sandwich(on, next, lunch):
... if lunch == 'sandwich':
... print 'delicious sandwich!'
... return next('dirt')
... else:
... return next(lunch)
>>> eat('soup')
eating soup
>>> eat('sandwich')
delicious sandwich!
eating dirt
>>> class Eater(object):
... def eat(self):
... print 'tastes like identity!'
... @classmethod
... def eat_class(cls):
... print 'let them eat cake!'
... @staticmethod
... def eat_static():
... print 'mmm, static cling'
... def eat_instance(self):
... print 'a moment in time'
>>> eater = Eater()
Multiple functions can be intercepted in one call to @advise, including
classmethods and staticmethods:
>>> @advise(Eater.eat, Eater.eat_class, Eater.eat_static, eater.eat_instance)
... def delicious(on, next):
... print 'delicious!'
... return next()
Normal method intercepted on the class:
>>> Eater().eat()
delicious!
tastes like identity!
Normal method intercepted on the instance:
>>> eater.eat_instance()
delicious!
a moment in time
Class method:
>>> Eater.eat_class()
delicious!
let them eat cake!
Static method:
>>> Eater.eat_static()
delicious!
mmm, static cling
Functions can be intercepted multiple times:
>>> @advise(Eater.eat)
... def intercept(on, next):
... print 'intercepted...AGAIN'
... return next()
>>> Eater().eat()
intercepted...AGAIN
delicious!
tastes like identity!
"""
hook = []
def hook_advice(join_point):
def intercept(*args, **kwargs):
return hook[0](on, join_point, *args, **kwargs)
intercept = functools.update_wrapper(intercept, join_point)
# Either a normal method or a class method?
if type(join_point) is types.MethodType:
# Class method intercept or instance intercept
if join_point.im_self:
on = join_point.im_self
# If we have hooked onto an instance method...
if type(on) is type:
def intercept(cls, *args, **kwargs):
return hook[0](cls, join_point, *args, **kwargs)
intercept = functools.update_wrapper(intercept, join_point)
intercept = classmethod(intercept)
else:
# Normal method, we curry "self" to make "next" uniform
def intercept(self, *args, **kwargs):
curry = functools.update_wrapper(
lambda *a, **kw: join_point(self, *a, **kw), join_point)
return hook[0](self, curry, *args, **kwargs)
intercept = functools.update_wrapper(intercept, join_point)
on = join_point.im_class
else:
# Static method or global function
on = sys.modules[join_point.__module__]
caller_globals = join_point.func_globals
name = join_point.__name__
# Global function
if caller_globals.get(name) is join_point:
caller_globals[name] = intercept
else:
# Probably a staticmethod, try to find the attached class
for on in caller_globals.values():
if getattr(on, name, None) is join_point:
intercept = staticmethod(intercept)
break
else:
raise ValueError('%s is not a global scope function and '
'could not be found in top-level classes'
% name)
name = join_point.__name__
setattr(on, name, intercept)
for join_point in join_points:
hook_advice(join_point)
def add_hook(func):
hook.append(func)
return func
return add_hook
if __name__ == '__main__':
import doctest
doctest.testmod()
|
Detailed examples are in the doctest, but using the classic "logging" advice example:
class A(object):
def a_function(self):
print 'a_function()'
@advise(A.a_function)
def logit(on, next, *args, **kwargs):
logging.debug('%r.%r(%r, %r)', on, next, args, kwargs)
return next(*args, **kwargs)