Welcome, guest | Sign In | My Account | Store | Cart

There are a couple of other AoP related recipes on ASPN. This one is designed to have a simple, elegant interface.

Python, 156 lines
  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)