Size: 56689
Comment: new decorator proposal
|
Size: 56765
Comment: Add syntax highlighting to all examples
|
Deletions are marked like this. | Additions are marked like this. |
Line 299: | Line 299: |
#!python | |
Line 1109: | Line 1110: |
#!python | |
Line 1261: | Line 1263: |
#!python | |
Line 1472: | Line 1475: |
#!python | |
Line 1633: | Line 1637: |
#!python | |
Line 1650: | Line 1655: |
#!python | |
Line 1683: | Line 1689: |
#!python | |
Line 1720: | Line 1726: |
#!python |
This page is meant to be a central repository of decorator code pieces, whether useful or not <wink>. It is NOT a page to discuss decorator syntax!
Feel free to add your suggestions. Please make sure example code conforms with PEP 8.
Contents
- Creating Well-Behaved Decorators / "Decorator decorator"
- Property Definition
- Memoize
- Cached Properties
- Retry
- Pseudo-currying
- Creating decorator with optional arguments
- Controllable DIY debug
- Easy adding methods to a class instance
- Counting function calls
- Alternate Counting function calls
- Generating Deprecation Warnings
- Smart deprecation warnings (with valid filenames, line numbers, etc.)
- Enable/Disable Decorators
- Easy Dump of Function Arguments
- Pre-/Post-Conditions
- Profiling/Coverage Analysis
- Line Tracing Individual Functions
- Synchronization
- Type Enforcement (accepts/returns)
- CGI method wrapper
- State Machine Implementaion
- C++/Java-keyword-like function decorators
- Different Decorator Forms
- Unimplemented function replacement
- Redirects stdout printing to python standard logging.
- Access control
- Events rising and handling
- Singleton
- Asynchronous Call
- Class method decorator using instance
Creating Well-Behaved Decorators / "Decorator decorator"
Note: This is only one recipe. Others include inheritance from a standard decorator (link?), the functools @wraps decorator, and a factory function such as Michele Simionato's decorator module which even preserves signature information.
1 def simple_decorator(decorator):
2 """This decorator can be used to turn simple functions
3 into well-behaved decorators, so long as the decorators
4 are fairly simple. If a decorator expects a function and
5 returns a function (no descriptors), and if it doesn't
6 modify function attributes or docstring, then it is
7 eligible to use this. Simply apply @simple_decorator to
8 your decorator and it will automatically preserve the
9 docstring and function attributes of functions to which
10 it is applied."""
11 def new_decorator(f):
12 g = decorator(f)
13 g.__name__ = f.__name__
14 g.__doc__ = f.__doc__
15 g.__dict__.update(f.__dict__)
16 return g
17 # Now a few lines needed to make simple_decorator itself
18 # be a well-behaved decorator.
19 new_decorator.__name__ = decorator.__name__
20 new_decorator.__doc__ = decorator.__doc__
21 new_decorator.__dict__.update(decorator.__dict__)
22 return new_decorator
23
24 #
25 # Sample Use:
26 #
27 @simple_decorator
28 def my_simple_logging_decorator(func):
29 def you_will_never_see_this_name(*args, **kwargs):
30 print 'calling %s' % func.__name__
31 return func(*args, **kwargs)
32 return you_will_never_see_this_name
33
34 @my_simple_logging_decorator
35 def double(x):
36 "Doubles a number"
37 return 2*x
38
39 assert double.__name__ == 'double'
40 assert double.__doc__ == 'Doubles a number'
41 print double(155)
Property Definition
These decorators provide a readable way to define properties:
1 import sys
2
3 def propget(func):
4 locals = sys._getframe(1).f_locals
5 name = func.__name__
6 prop = locals.get(name)
7 if not isinstance(prop, property):
8 prop = property(func, doc=func.__doc__)
9 else:
10 doc = prop.__doc__ or func.__doc__
11 prop = property(func, prop.fset, prop.fdel, doc)
12 return prop
13
14 def propset(func):
15 locals = sys._getframe(1).f_locals
16 name = func.__name__
17 prop = locals.get(name)
18 if not isinstance(prop, property):
19 prop = property(None, func, doc=func.__doc__)
20 else:
21 doc = prop.__doc__ or func.__doc__
22 prop = property(prop.fget, func, prop.fdel, doc)
23 return prop
24
25 def propdel(func):
26 locals = sys._getframe(1).f_locals
27 name = func.__name__
28 prop = locals.get(name)
29 if not isinstance(prop, property):
30 prop = property(None, None, func, doc=func.__doc__)
31 else:
32 prop = property(prop.fget, prop.fset, func, prop.__doc__)
33 return prop
34
35 # These can be used like this:
36
37 class Example(object):
38
39 @propget
40 def myattr(self):
41 return self._half * 2
42
43 @propset
44 def myattr(self, value):
45 self._half = value / 2
46
47 @propdel
48 def myattr(self):
49 del self._half
Here's a way that doesn't require any new decorators:
Yet another property decorator:
1 def property(function):
2 keys = 'fget', 'fset', 'fdel'
3 func_locals = {'doc':function.__doc__}
4 def probe_func(frame, event, arg):
5 if event == 'return':
6 locals = frame.f_locals
7 func_locals.update(dict((k, locals.get(k)) for k in keys))
8 sys.settrace(None)
9 return probe_func
10 sys.settrace(probe_func)
11 function()
12 return property(**func_locals)
13
14 #====== Example =======================================================
15
16 from math import radians, degrees, pi
17
18 class Angle(object):
19 def __init__(self, rad):
20 self._rad = rad
21
22 @property
23 def rad():
24 '''The angle in radians'''
25 def fget(self):
26 return self._rad
27 def fset(self, angle):
28 if isinstance(angle, Angle):
29 angle = angle.rad
30 self._rad = float(angle)
31
32 @property
33 def deg():
34 '''The angle in degrees'''
35 def fget(self):
36 return degrees(self._rad)
37 def fset(self, angle):
38 if isinstance(angle, Angle):
39 angle = angle.deg
40 self._rad = radians(angle)
Memoize
Here's a memoizing class.
1 class memoized(object):
2 """Decorator that caches a function's return value each time it is called.
3 If called later with the same arguments, the cached value is returned, and
4 not re-evaluated.
5 """
6 def __init__(self, func):
7 self.func = func
8 self.cache = {}
9 def __call__(self, *args):
10 try:
11 return self.cache[args]
12 except KeyError:
13 value = self.func(*args)
14 self.cache[args] = value
15 return value
16 except TypeError:
17 # uncachable -- for instance, passing a list as an argument.
18 # Better to not cache than to blow up entirely.
19 return self.func(*args)
20 def __repr__(self):
21 """Return the function's docstring."""
22 return self.func.__doc__
23 def __get__(self, obj, objtype):
24 """Support instance methods."""
25 return functools.partial(self.__call__, obj)
26
27 @memoized
28 def fibonacci(n):
29 "Return the nth fibonacci number."
30 if n in (0, 1):
31 return n
32 return fibonacci(n-1) + fibonacci(n-2)
33
34 print fibonacci(12)
Cached Properties
1 #
2 # © 2011 Christopher Arndt, MIT License
3 #
4
5 import time
6
7 class cached_property(object):
8 """Decorator for read-only properties evaluated only once within TTL period.
9
10 It can be used to created a cached property like this::
11
12 import random
13
14 # the class containing the property must be a new-style class
15 class MyClass(object):
16 # create property whose value is cached for ten minutes
17 @cached_property(ttl=600)
18 def randint(self):
19 # will only be evaluated every 10 min. at maximum.
20 return random.randint(0, 100)
21
22 The value is cached in the '_cache' attribute of the object instance that
23 has the property getter method wrapped by this decorator. The '_cache'
24 attribute value is a dictionary which has a key for every property of the
25 object which is wrapped by this decorator. Each entry in the cache is
26 created only when the property is accessed for the first time and is a
27 two-element tuple with the last computed property value and the last time
28 it was updated in seconds since the epoch.
29
30 The default time-to-live (TTL) is 300 seconds (5 minutes). Set the TTL to
31 zero for the cached value to never expire.
32
33 To expire a cached property value manually just do::
34
35 del instance._cache[<property name>]
36
37 """
38 def __init__(self, ttl=300):
39 self.ttl = ttl
40
41 def __call__(self, fget, doc=None):
42 self.fget = fget
43 self.__doc__ = doc or fget.__doc__
44 self.__name__ = fget.__name__
45 self.__module__ = fget.__module__
46 return self
47
48 def __get__(self, inst, owner):
49 now = time.time()
50 try:
51 value, last_update = inst._cache[self.__name__]
52 if self.ttl > 0 and now - last_update > self.ttl:
53 raise AttributeError
54 except (KeyError, AttributeError):
55 value = self.fget(inst)
56 try:
57 cache = inst._cache
58 except AttributeError:
59 cache = inst._cache = {}
60 cache[self.__name__] = (value, now)
61 return value
Retry
Call a function which returns True/False to indicate success or failure. On failure, wait, and try the function again. On repeated failures, wait longer between each successive attempt. If the decorator runs out of attempts, then it gives up and returns False, but you could just as easily raise some exception.
1 import time
2
3 # Retry decorator with exponential backoff
4 def retry(tries, delay=3, backoff=2):
5 """Retries a function or method until it returns True.
6
7 delay sets the initial delay, and backoff sets how much the delay should
8 lengthen after each failure. backoff must be greater than 1, or else it
9 isn't really a backoff. tries must be at least 0, and delay greater than
10 0."""
11
12 if backoff <= 1:
13 raise ValueError("backoff must be greater than 1")
14
15 tries = math.floor(tries)
16 if tries < 0:
17 raise ValueError("tries must be 0 or greater")
18
19 if delay <= 0:
20 raise ValueError("delay must be greater than 0")
21
22 def deco_retry(f):
23 def f_retry(*args, **kwargs):
24 mtries, mdelay = tries, delay # make mutable
25
26 rv = f(*args, **kwargs) # first attempt
27 while mtries > 0:
28 if rv == True: # Done on success
29 return True
30
31 mtries -= 1 # consume an attempt
32 time.sleep(mdelay) # wait...
33 mdelay *= backoff # make future wait longer
34
35 rv = f(*args, **kwargs) # Try again
36
37 return False # Ran out of tries :-(
38
39 return f_retry # true decorator -> decorated function
40 return deco_retry # @retry(arg[, ...]) -> true decorator
Pseudo-currying
1 class curried(object):
2 """
3 Decorator that returns a function that keeps returning functions
4 until all arguments are supplied; then the original function is
5 evaluated.
6 """
7
8 def __init__(self, func, *a):
9 self.func = func
10 self.args = a
11 def __call__(self, *a):
12 args = self.args + a
13 if len(args) < self.func.func_code.co_argcount:
14 return curried(self.func, *args)
15 else:
16 return self.func(*args)
17
18
19 @curried
20 def add(a, b):
21 return a + b
22
23 add1 = add(1)
24
25 print add1(2)
Creating decorator with optional arguments
1 import functools, inspect
2
3 def decorator(func):
4 """ Allow to use decorator either with arguments or not. """
5
6 def isFuncArg(*args, **kw):
7 return len(args) == 1 and len(kw) == 0 and (
8 inspect.isfunction(args[0]) or isinstance(args[0], type))
9
10 if isinstance(func, type):
11 def class_wrapper(*args, **kw):
12 if isFuncArg(*args, **kw):
13 return func()(*args, **kw) # create class before usage
14 return func(*args, **kw)
15 class_wrapper.__name__ = func.__name__
16 class_wrapper.__module__ = func.__module__
17 return class_wrapper
18
19 @functools.wraps(func)
20 def func_wrapper(*args, **kw):
21 if isFuncArg(*args, **kw):
22 return func(*args, **kw)
23
24 def functor(userFunc):
25 return func(userFunc, *args, **kw)
26
27 return functor
28
29 return func_wrapper
Example:
1 @decorator
2 def apply(func, *args, **kw):
3 return func(*args, **kw)
4
5 @decorator
6 class apply:
7 def __init__(self, *args, **kw):
8 self.args = args
9 self.kw = kw
10
11 def __call__(self, func):
12 return func(*self.args, **self.kw)
13
14 #
15 # Usage in both cases:
16 #
17 @apply
18 def test():
19 return 'test'
20
21 assert test == 'test'
22
23 @apply(2, 3)
24 def test(a, b):
25 return a + b
26
27 assert test == 5
Note: There is only one drawback: wrapper checks its arguments for single function or class. To avoid wrong behavior you can use keyword arguments instead of positional, e.g.:
1 @decorator
2 def my_property(getter, *, setter = None, deleter = None, doc = None):
3 return property(getter, setter, deleter, doc)
Controllable DIY debug
(Other hooks could be similarly added. Docstrings and exceptions are left out for simplicity of demonstration.)
1 import sys
2
3 WHAT_TO_DEBUG = set(['io', 'core']) # change to what you need
4
5 class debug:
6 """ Decorator which helps to control what aspects of a program to debug
7 on per-function basis. Aspects are provided as list of arguments.
8 It DOESN'T slowdown functions which aren't supposed to be debugged.
9 """
10 def __init__(self, aspects=None):
11 self.aspects = set(aspects)
12
13 def __call__(self, f):
14 if self.aspects & WHAT_TO_DEBUG:
15 def newf(*args, **kwds):
16 print >> sys.stderr, f.func_name, args, kwds
17 f_result = f(*args, **kwds)
18 print >> sys.stderr, f.func_name, "returned", f_result
19 return f_result
20 newf.__doc__ = f.__doc__
21 return newf
22 else:
23 return f
24
25 @debug(['io'])
26 def prn(x):
27 print x
28
29 @debug(['core'])
30 def mult(x, y):
31 return x * y
32
33 prn(mult(2, 2))
Easy adding methods to a class instance
Credits to John Roth.
1 class Foo:
2 def __init__(self):
3 self.x = 42
4
5 foo = Foo()
6
7 def addto(instance):
8 def decorator(f):
9 import types
10 f = types.MethodType(f, instance, instance.__class__)
11 setattr(instance, f.func_name, f)
12 return f
13 return decorator
14
15 @addto(foo)
16 def print_x(self):
17 print self.x
18
19 # foo.print_x() would print "42"
Counting function calls
1 class countcalls(object):
2 "Decorator that keeps track of the number of times a function is called."
3
4 __instances = {}
5
6 def __init__(self, f):
7 self.__f = f
8 self.__numcalls = 0
9 countcalls.__instances[f] = self
10
11 def __call__(self, *args, **kwargs):
12 self.__numcalls += 1
13 return self.__f(*args, **kwargs)
14
15 @staticmethod
16 def count(f):
17 "Return the number of times the function f was called."
18 return countcalls.__instances[f].__numcalls
19
20 @staticmethod
21 def counts():
22 "Return a dict of {function: # of calls} for all registered functions."
23 return dict([(f, countcalls.count(f)) for f in countcalls.__instances])
Alternate Counting function calls
1 class countcalls(object):
2 "Decorator that keeps track of the number of times a function is called."
3
4 __instances = {}
5
6 def __init__(self, f):
7 self.__f = f
8 self.__numcalls = 0
9 countcalls.__instances[f] = self
10
11 def __call__(self, *args, **kwargs):
12 self.__numcalls += 1
13 return self.__f(*args, **kwargs)
14
15 def count(self):
16 "Return the number of times the function f was called."
17 return countcalls.__instances[self.__f].__numcalls
18
19 @staticmethod
20 def counts():
21 "Return a dict of {function: # of calls} for all registered functions."
22 return dict([(f.__name__, countcalls.__instances[f].__numcalls) for f in countcalls.__instances])
23
24 #example
25
26 @countcalls
27 def f():
28 print 'f called'
29
30 @countcalls
31 def g():
32 print 'g called'
33
34 f()
35 f()
36 f()
37 print f.count() # prints 3
38 print countcalls.counts() # same as f.counts() or g.counts()
39 g()
40 print g.count() # prints 1
Generating Deprecation Warnings
1 import warnings
2
3 def deprecated(func):
4 """This is a decorator which can be used to mark functions
5 as deprecated. It will result in a warning being emitted
6 when the function is used."""
7 def new_func(*args, **kwargs):
8 warnings.warn("Call to deprecated function %s." % func.__name__,
9 category=DeprecationWarning)
10 return func(*args, **kwargs)
11 new_func.__name__ = func.__name__
12 new_func.__doc__ = func.__doc__
13 new_func.__dict__.update(func.__dict__)
14 return new_func
15
16 # === Examples of use ===
17
18 @deprecated
19 def some_old_function(x,y):
20 return x + y
21
22 class SomeClass:
23 @deprecated
24 def some_old_method(self, x,y):
25 return x + y
Smart deprecation warnings (with valid filenames, line numbers, etc.)
1 import warnings
2 import functools
3
4
5 def deprecated(func):
6 """This is a decorator which can be used to mark functions
7 as deprecated. It will result in a warning being emitted
8 when the function is used."""
9
10 @functools.wraps(func)
11 def new_func(*args, **kwargs):
12 warnings.warn_explicit(
13 "Call to deprecated function %(funcname)s." % {
14 'funcname': func.__name__,
15 },
16 category=DeprecationWarning,
17 filename=func.func_code.co_filename,
18 lineno=func.func_code.co_firstlineno + 1
19 )
20 return func(*args, **kwargs)
21 return new_func
22
23
24 ## Usage examples ##
25 @deprecated
26 def my_func():
27 pass
28
29 @other_decorators_must_be_upper
30 @deprecated
31 def my_func():
32 pass
Enable/Disable Decorators
1 def unchanged(func):
2 "This decorator doesn't add any behavior"
3 return func
4
5 def disabled(func):
6 "This decorator disables the provided function, and does nothing"
7 def empty_func(*args,**kargs):
8 pass
9 return empty_func
10
11 # define this as equivalent to unchanged, for nice symmetry with disabled
12 enabled = unchanged
13
14 #
15 # Sample use
16 #
17
18 global_enable_flag = True
19
20 state = enabled if global_enable_flag else disabled
21 @state
22 def special_function_foo():
23 print "function was enabled"
Easy Dump of Function Arguments
1 def dump_args(func):
2 "This decorator dumps out the arguments passed to a function before calling it"
3 argnames = func.func_code.co_varnames[:func.func_code.co_argcount]
4 fname = func.func_name
5 def echo_func(*args,**kwargs):
6 print fname, ":", ', '.join(
7 '%s=%r' % entry
8 for entry in zip(argnames,args) + kwargs.items())
9 return func(*args, **kwargs)
10 return echo_func
11
12 @dump_args
13 def f1(a,b,c):
14 print a + b + c
15
16 f1(1, 2, 3)
Pre-/Post-Conditions
1 """
2 Provide pre-/postconditions as function decorators.
3
4 Example usage:
5
6 >>> def in_ge20(inval):
7 ... assert inval >= 20, 'Input value < 20'
8 ...
9 >>> def out_lt30(retval, inval):
10 ... assert retval < 30, 'Return value >= 30'
11 ...
12 >>> @precondition(in_ge20)
13 ... @postcondition(out_lt30)
14 ... def inc(value):
15 ... return value + 1
16 ...
17 >>> inc(5)
18 Traceback (most recent call last):
19 ...
20 AssertionError: Input value < 20
21 >>> inc(29)
22 Traceback (most recent call last):
23 ...
24 AssertionError: Return value >= 30
25 >>> inc(20)
26 21
27
28 You can define as many pre-/postconditions for a function as you
29 like. It is also possible to specify both types of conditions at once:
30
31 >>> @conditions(in_ge20, out_lt30)
32 ... def add1(value):
33 ... return value + 1
34 ...
35 >>> add1(5)
36 Traceback (most recent call last):
37 ...
38 AssertionError: Input value < 20
39
40 An interesting feature is the ability to prevent the creation of
41 pre-/postconditions at function definition time. This makes it
42 possible to use conditions for debugging and then switch them off for
43 distribution.
44
45 >>> debug = False
46 >>> @precondition(in_ge20, debug)
47 ... def dec(value):
48 ... return value - 1
49 ...
50 >>> dec(5)
51 4
52 """
53
54 __all__ = ['precondition', 'postcondition', 'conditions']
55
56 DEFAULT_ON = True
57
58 def precondition(precondition, use_conditions=DEFAULT_ON):
59 return conditions(precondition, None, use_conditions)
60
61 def postcondition(postcondition, use_conditions=DEFAULT_ON):
62 return conditions(None, postcondition, use_conditions)
63
64 class conditions(object):
65 __slots__ = ('__precondition', '__postcondition')
66
67 def __init__(self, pre, post, use_conditions=DEFAULT_ON):
68 if not use_conditions:
69 pre, post = None, None
70
71 self.__precondition = pre
72 self.__postcondition = post
73
74 def __call__(self, function):
75 # combine recursive wrappers (@precondition + @postcondition == @conditions)
76 pres = set((self.__precondition,))
77 posts = set((self.__postcondition,))
78
79 # unwrap function, collect distinct pre-/post conditions
80 while type(function) is FunctionWrapper:
81 pres.add(function._pre)
82 posts.add(function._post)
83 function = function._func
84
85 # filter out None conditions and build pairs of pre- and postconditions
86 conditions = map(None, filter(None, pres), filter(None, posts))
87
88 # add a wrapper for each pair (note that 'conditions' may be empty)
89 for pre, post in conditions:
90 function = FunctionWrapper(pre, post, function)
91
92 return function
93
94 class FunctionWrapper(object):
95 def __init__(self, precondition, postcondition, function):
96 self._pre = precondition
97 self._post = postcondition
98 self._func = function
99
100 def __call__(self, *args, **kwargs):
101 precondition = self._pre
102 postcondition = self._post
103
104 if precondition:
105 precondition(*args, **kwargs)
106 result = self._func(*args, **kwargs)
107 if postcondition:
108 postcondition(result, *args, **kwargs)
109 return result
110
111 def __test():
112 import doctest
113 doctest.testmod()
114
115 if __name__ == "__main__":
116 __test()
Profiling/Coverage Analysis
The code and examples are a bit longish, so I'll include a link instead: http://mg.pov.lt/blog/profiling.html
Line Tracing Individual Functions
I cobbled this together from the trace module. It allows you to decorate individual functions so their lines are traced. I think it works out to be a slightly smaller hammer than running the trace module and trying to pare back what it traces using exclusions.
1 import sys
2 import os
3 import linecache
4
5 def trace(f):
6 def globaltrace(frame, why, arg):
7 if why == "call":
8 return localtrace
9 return None
10
11 def localtrace(frame, why, arg):
12 if why == "line":
13 # record the file name and line number of every trace
14 filename = frame.f_code.co_filename
15 lineno = frame.f_lineno
16
17 bname = os.path.basename(filename)
18 print "%s(%d): %s" % (bname, lineno,
19 linecache.getline(filename, lineno)),
20 return localtrace
21
22 def _f(*args, **kwds):
23 sys.settrace(globaltrace)
24 result = f(*args, **kwds)
25 sys.settrace(None)
26 return result
27
28 return _f
Synchronization
Synchronize two (or more) functions on a given lock.
1 def synchronized(lock):
2 """ Synchronization decorator. """
3
4 def wrap(f):
5 def new_function(*args, **kw):
6 lock.acquire()
7 try:
8 return f(*args, **kw)
9 finally:
10 lock.release()
11 return new_function
12 return wrap
13
14 # Example usage:
15
16 from threading import Lock
17 my_lock = Lock()
18
19 @synchronized(my_lock)
20 def critical1(*args):
21 # Interesting stuff goes here.
22 pass
23
24 @synchronized(my_lock)
25 def critical2(*args):
26 # Other interesting stuff goes here.
27 pass
Type Enforcement (accepts/returns)
Provides various degrees of type enforcement for function parameters and return values.
1 """
2 One of three degrees of enforcement may be specified by passing
3 the 'debug' keyword argument to the decorator:
4 0 -- NONE: No type-checking. Decorators disabled.
5 1 -- MEDIUM: Print warning message to stderr. (Default)
6 2 -- STRONG: Raise TypeError with message.
7 If 'debug' is not passed to the decorator, the default level is used.
8
9 Example usage:
10 >>> NONE, MEDIUM, STRONG = 0, 1, 2
11 >>>
12 >>> @accepts(int, int, int)
13 ... @returns(float)
14 ... def average(x, y, z):
15 ... return (x + y + z) / 2
16 ...
17 >>> average(5.5, 10, 15.0)
18 TypeWarning: 'average' method accepts (int, int, int), but was given
19 (float, int, float)
20 15.25
21 >>> average(5, 10, 15)
22 TypeWarning: 'average' method returns (float), but result is (int)
23 15
24
25 Needed to cast params as floats in function def (or simply divide by 2.0).
26
27 >>> TYPE_CHECK = STRONG
28 >>> @accepts(int, debug=TYPE_CHECK)
29 ... @returns(int, debug=TYPE_CHECK)
30 ... def fib(n):
31 ... if n in (0, 1): return n
32 ... return fib(n-1) + fib(n-2)
33 ...
34 >>> fib(5.3)
35 Traceback (most recent call last):
36 ...
37 TypeError: 'fib' method accepts (int), but was given (float)
38
39 """
40 import sys
41
42 def accepts(*types, **kw):
43 """ Function decorator. Checks that inputs given to decorated function
44 are of the expected type.
45
46 Parameters:
47 types -- The expected types of the inputs to the decorated function.
48 Must specify type for each parameter.
49 kw -- Optional specification of 'debug' level (this is the only valid
50 keyword argument, no other should be given).
51 debug = ( 0 | 1 | 2 )
52
53 """
54 if not kw:
55 # default level: MEDIUM
56 debug = 1
57 else:
58 debug = kw['debug']
59 try:
60 def decorator(f):
61 def newf(*args):
62 if debug == 0:
63 return f(*args)
64 assert len(args) == len(types)
65 argtypes = tuple(map(type, args))
66 if argtypes != types:
67 msg = info(f.__name__, types, argtypes, 0)
68 if debug == 1:
69 print >> sys.stderr, 'TypeWarning: ', msg
70 elif debug == 2:
71 raise TypeError, msg
72 return f(*args)
73 newf.__name__ = f.__name__
74 return newf
75 return decorator
76 except KeyError, key:
77 raise KeyError, key + "is not a valid keyword argument"
78 except TypeError, msg:
79 raise TypeError, msg
80
81
82 def returns(ret_type, **kw):
83 """ Function decorator. Checks that return value of decorated function
84 is of the expected type.
85
86 Parameters:
87 ret_type -- The expected type of the decorated function's return value.
88 Must specify type for each parameter.
89 kw -- Optional specification of 'debug' level (this is the only valid
90 keyword argument, no other should be given).
91 debug=(0 | 1 | 2)
92
93 """
94 try:
95 if not kw:
96 # default level: MEDIUM
97 debug = 1
98 else:
99 debug = kw['debug']
100 def decorator(f):
101 def newf(*args):
102 result = f(*args)
103 if debug == 0:
104 return result
105 res_type = type(result)
106 if res_type != ret_type:
107 msg = info(f.__name__, (ret_type,), (res_type,), 1)
108 if debug == 1:
109 print >> sys.stderr, 'TypeWarning: ', msg
110 elif debug == 2:
111 raise TypeError, msg
112 return result
113 newf.__name__ = f.__name__
114 return newf
115 return decorator
116 except KeyError, key:
117 raise KeyError, key + "is not a valid keyword argument"
118 except TypeError, msg:
119 raise TypeError, msg
120
121 def info(fname, expected, actual, flag):
122 """ Convenience function returns nicely formatted error/warning msg. """
123 format = lambda types: ', '.join([str(t).split("'")[1] for t in types])
124 expected, actual = format(expected), format(actual)
125 msg = "'%s' method " % fname \
126 + ("accepts", "returns")[flag] + " (%s), but " % expected\
127 + ("was given", "result is")[flag] + " (%s)" % actual
128 return msg
CGI method wrapper
Handles HTML boilerplate at top and bottom of pages returned from CGI methods. Works with the cgi module. Now your request handlers can just output the interesting HTML, and let the decorator deal with all the top and bottom clutter.
(Note: the exception handler eats all exceptions, which in CGI is no big loss, since the program runs in its separate subprocess. At least here, the exception contents will be written to the output page.)
1 class CGImethod(object):
2 def __init__(self, title):
3 self.title = title
4 def __call__(self, fn):
5 def wrapped_fn(*args):
6 print "Content-Type: text/html\n\n"
7 print "<HTML>"
8 print "<HEAD><TITLE>%s</TITLE></HEAD>" % self.title
9 print "<BODY>"
10 try:
11 fn(*args)
12 except Exception, e:
13 print
14 print e
15 print
16 print "</BODY></HTML>"
17
18 return wrapped_fn
19
20 @CGImethod("Hello with Decorator")
21 def say_hello():
22 print '<h1>Hello from CGI-Land</h1>'
State Machine Implementaion
A much improved version of decorators for implementing state machines, too long to show here, is at State Machine via Decorators
This example uses Decorators to facilitate the implementation of a state machine in Python. Decorators are used to specify which methods are the event handlers for the class. In this example, actions are associated with the transitions, but it is possible with a little consideration to associate actions with states instead.
The example defines a class, MyMachine that is a state machine. Multiple instances of the class may be instantiated with each maintaining its own state. A class also may have multiple states. Here I've used gstate and tstate.
The code in the imported statedefn file gets a bit hairy, but you may not need to delve into it for your application.
1 # State Machine example Program
2
3 from statedefn import *
4
5 class MyMachine(object):
6
7 # Create Statedefn object for each state you need to keep track of.
8 # the name passed to the constructor becomes a StateVar member of the current class.
9 # i.e. if my_obj is a MyMachine object, my_obj.gstate maintains the current gstate
10 gstate = StateTable("gstate")
11 tstate = StateTable("turtle")
12
13 def __init__(self, name):
14 # must call init method of class's StateTable object. to initialize state variable
15 self.gstate.initialize(self)
16 self.tstate.initialize(self)
17 self.mname = name
18 self.a_count = 0
19 self.b_count = 0
20 self.c_count = 0
21
22 # Decorate the Event Handler virtual functions -note gstate parameter
23 @event_handler(gstate)
24 def event_a(self): pass
25 @event_handler(gstate)
26 def event_b(self): pass
27 @event_handler(gstate)
28 def event_c(self, val): pass
29
30 @event_handler(tstate)
31 def toggle(self): pass
32
33
34 # define methods to handle events.
35 def _event_a_hdlr1(self):
36 print "State 1, event A"
37 self.a_count += 1
38 def _event_b_hdlr1(self):
39 print "State 1, event B"
40 self.b_count += 1
41 def _event_c_hdlr1(self, val):
42 print "State 1, event C"
43 self.c_count += 3*val
44
45 def _event_a_hdlr2(self):
46 print "State 2, event A"
47 self.a_count += 10
48 # here we brute force the tstate to on, leave & enter functions called if state changes.
49 # turtle is object's state variable for tstate, comes from constructor argument
50 self.turtle.set_state(self, self._t_on)
51 def _event_b_hdlr2(self):
52 print "State 2, event B"
53 self.b_count += 10
54 def _event_c_hdlr2(self, val):
55 print "State 2, event C"
56 self.c_count += 2*val
57
58 def _event_a_hdlr3(self):
59 self.a_count += 100
60 print "State 3, event A"
61 def _event_b_hdlr3(self):
62 print "State 3, event B"
63 self.b_count += 100
64 # we decide here we want to go to state 2, overrrides spec in state table below.
65 # transition to next_state is made after the method exits.
66 self.gstate.next_state = self._state2
67 def _event_c_hdlr3(self, val):
68 print "State 3, event C"
69 self.c_count += 5*val
70
71 # Associate the handlers with a state. The first argument is a list of methods.
72 # One method for each event_handler decorated function of gstate. Order of methods
73 # in the list correspond to order in which the Event Handlers were declared.
74 # Second arg is the name of the state. Third argument is to be come a list of the
75 # next states.
76 # The first state created becomes the initial state.
77 _state1 = gstate.state("One", (_event_a_hdlr1, _event_b_hdlr1, _event_c_hdlr1),
78 ("Two", "Three", None))
79 _state2 = gstate.state("Two", (_event_a_hdlr2, _event_b_hdlr2, _event_c_hdlr2),
80 ("Three", None, "One"))
81 _state3 = gstate.state("Three",(_event_a_hdlr3, _event_b_hdlr3, _event_c_hdlr3),
82 (None, "One", "Two"))
83
84
85 # Declare a function that will be called when entering a new gstate.
86 # Can also declare a leave function using @on_leave_function(gstate)
87 @on_enter_function(gstate)
88 def _enter_gstate(self):
89 print "entering state ", self.gstate.name() , "of ", self.mname
90 @on_leave_function(tstate)
91 def _leave_tstate(self):
92 print "leaving state ", self.turtle.name() , "of ", self.mname
93
94
95 def _toggle_on(self):
96 print "Toggle On"
97
98 def _toggle_off(self):
99 print "Toggle Off"
100
101 _t_off = tstate.state("Off", [_toggle_on],
102 ["On"])
103 _t_on = tstate.state("On", [_toggle_off],
104 ["Off"])
105
106
107 def main():
108 big_machine = MyMachine("big")
109 lil_machine = MyMachine("lil")
110
111 big_machine.event_a()
112 lil_machine.event_a()
113 big_machine.event_a()
114 lil_machine.event_a()
115 big_machine.event_b()
116 lil_machine.event_b()
117 big_machine.event_c(4)
118 lil_machine.event_c(2)
119 big_machine.event_c(1)
120 lil_machine.event_c(3)
121 big_machine.event_b()
122 lil_machine.event_b()
123 big_machine.event_a()
124 lil_machine.event_a()
125 big_machine.event_a()
126
127 big_machine.toggle()
128 big_machine.toggle()
129 big_machine.toggle()
130
131 lil_machine.event_a()
132 big_machine.event_b()
133 lil_machine.event_b()
134 big_machine.event_c(3)
135 big_machine.event_a()
136 lil_machine.event_c(2)
137 lil_machine.event_a()
138 big_machine.event_b()
139 lil_machine.event_b()
140 big_machine.event_c(7)
141 lil_machine.event_c(1)
142
143 print "Event A count ", big_machine.a_count
144 print "Event B count ", big_machine.b_count
145 print "Event C count ", big_machine.c_count
146 print "LilMachine C count ", lil_machine.c_count
147
148 main()
And now the imported statedefn.py
1 #
2 # Support for State Machines. ref - Design Patterns by GoF
3 # Many of the methods in these classes get called behind the scenes.
4 #
5 # Notable exceptions are methods of the StateVar class.
6 #
7 # See example programs for how this module is intended to be used.
8 #
9 class StateMachineError(Exception):
10 def __init__(self, args = None):
11 self.args = args
12
13 class StateVar(object):
14 def __init__(self, initial_state):
15 self._current_state = initial_state
16 self.next_state = initial_state # publicly settable in an event handling routine.
17
18 def set_state(self, owner, new_state):
19 '''
20 Forces a state change to new_state
21 '''
22 self.next_state = new_state
23 self.__to_next_state(owner)
24
25 def __to_next_state(self, owner):
26 '''
27 The low-level state change function which calls leave state & enter state functions as
28 needed.
29
30 LeaveState and EnterState functions are called as needed when state transitions.
31 '''
32 if self.next_state is not self._current_state:
33 if hasattr(self._current_state, "leave"):
34 self._current_state.leave(owner)
35 elif hasattr(self, "leave"):
36 self.leave(owner)
37 self._current_state = self.next_state
38 if hasattr(self._current_state, "enter"):
39 self._current_state.enter(owner)
40 elif hasattr(self, "enter"):
41 self.enter(owner)
42
43 def __fctn(self, func_name):
44 '''
45 Returns the owning class's method for handling an event for the current state.
46 This method not for public consumption.
47 '''
48 vf = self._current_state.get_fe(func_name)
49 return vf
50
51 def name(self):
52 '''
53 Returns the current state name.
54 '''
55 return self._current_state.name
56
57 class STState(object):
58 def __init__(self, state_name):
59 self.name = state_name
60 self.fctn_dict = {}
61
62 def set_events(self, event_list, event_hdlr_list, next_states):
63 dictionary = self.fctn_dict
64 if not next_states:
65 def set_row(event, method):
66 dictionary[event] = [method, None]
67 map(set_row, event_list, event_hdlr_list)
68 else:
69 def set_row2(event, method, next_state):
70 dictionary[event] = [method, next_state]
71 map(set_row2, event_list, event_hdlr_list, next_states)
72 self.fctn_dict = dictionary
73
74 def get_fe(self, fctn_name):
75 return self.fctn_dict[fctn_name]
76
77 def map_next_states(self, state_dict):
78 ''' Changes second dict value from name of state to actual state '''
79 for de in self.fctn_dict.values():
80 next_state_name = de[1]
81 if next_state_name:
82 if next_state_name in state_dict:
83 de[1] = state_dict[next_state_name]
84 else:
85 raise StateMachineError('Invalid Name for next state: %s' % next_state_name)
86
87
88 class StateTable(object):
89 '''
90 Magical class to define a state machine, with the help of several decorator functions
91 which follow.
92 '''
93 def __init__(self, declname):
94 self.machine_var = declname
95 self._initial_state = None
96 self._state_list = {}
97 self._event_list = []
98 self.need_initialize = 1
99
100 def initialize(self, parent):
101 '''
102 Initializes the parent class's state variable for this StateTable class.
103 Must call this method in the parent' object's __init__ method. You can have
104 Multiple state machines within a parent class. Call this method for each
105 '''
106 statevar= StateVar(self._initial_state)
107 setattr(parent, self.machine_var, statevar)
108 if hasattr(self, "enter"):
109 statevar.enter = self.enter
110 if hasattr(self, "leave"):
111 statevar.leave = self.leave
112 #Magic happens here - in the 'next state' table, translate names into state objects.
113 if self.need_initialize:
114 for xstate in list(self._state_list.values()):
115 xstate.map_next_states(self._state_list)
116 self.need_initialize = 0
117
118 def def_state(self, event_hdlr_list, name):
119 '''
120 This is used to define a state. the event handler list is a list of functions that
121 are called for corresponding events. name is the name of the state.
122 '''
123 state_table_row = STState(name)
124 if len(event_hdlr_list) != len(self._event_list):
125 raise StateMachineError('Mismatch between number of event handlers and the methods specified for the state.')
126
127 state_table_row.set_events(self._event_list, event_hdlr_list, None)
128
129 if self._initial_state is None:
130 self._initial_state = state_table_row
131 self._state_list[name] = state_table_row
132 return state_table_row
133
134 def state(self, name, event_hdlr_list, next_states):
135 state_table_row = STState(name)
136 if len(event_hdlr_list) != len(self._event_list):
137 raise StateMachineError('Mismatch between number of event handlers and the methods specified for the state.')
138 if next_states is not None and len(next_states) != len(self._event_list):
139 raise StateMachineError('Mismatch between number of event handlers and the next states specified for the state.')
140
141 state_table_row.set_events(self._event_list, event_hdlr_list, next_states)
142
143 if self._initial_state is None:
144 self._initial_state = state_table_row
145 self._state_list[name] = state_table_row
146 return state_table_row
147
148 def __add_ev_hdlr(self, func_name):
149 '''
150 Informs the class of an event handler to be added. We just need the name here. The
151 function name will later be associated with one of the functions in a list when a state is defined.
152 '''
153 self._event_list.append(func_name)
154
155 # Decorator functions ...
156 def event_handler(state_class):
157 '''
158 Declare a method that handles a type of event.
159 '''
160 def wrapper(func):
161 state_class._StateTable__add_ev_hdlr(func.__name__)
162 def obj_call(self, *args, **keywords):
163 state_var = getattr(self, state_class.machine_var)
164 funky, next_state = state_var._StateVar__fctn(func.__name__)
165 if next_state is not None:
166 state_var.next_state = next_state
167 rv = funky(self, *args, **keywords)
168 state_var._StateVar__to_next_state(self)
169 return rv
170 return obj_call
171 return wrapper
172
173 def on_enter_function(state_class):
174 '''
175 Declare that this method should be called whenever a new state is entered.
176 '''
177 def wrapper(func):
178 state_class.enter = func
179 return func
180 return wrapper
181
182 def on_leave_function(state_class):
183 '''
184 Declares that this method should be called whenever leaving a state.
185 '''
186 def wrapper(func):
187 state_class.leave = func
188 return func
189 return wrapper
C++/Java-keyword-like function decorators
@abstractMethod, @deprecatedMethod, @privateMethod, @protectedMethod, @raises, @parameterTypes, @returnType
The annotations provide run-time type checking and an alternative way to document code.
The code and documentation are long, so I offer a link: http://fightingquaker.com/pyanno/
Different Decorator Forms
There are operational differences between:
- Decorator with no arguments
- Decorator with arguments
- Decorator with wrapped class instance awareness
This example demonstrates the operational differences between the three using a skit taken from Episode 22: Bruces.
1 from sys import stdout,stderr
2 from pdb import set_trace as bp
3
4 class DecoTrace(object):
5 '''
6 Decorator class with no arguments
7
8 This can only be used for functions or methods where the instance
9 is not necessary
10
11 '''
12
13 def __init__(self, f):
14 self.f = f
15
16 def _showargs(self, *fargs, **kw):
17 print >> stderr, 'T: enter %s with args=%s, kw=%s' % (self.f.__name__, str(fargs), str(kw))
18
19 def _aftercall(self, status):
20 print >> stderr, 'T: exit %s with status=%s' % (self.f.__name__, str(status))
21
22 def __call__(self, *fargs, **kw):
23 '''pass just function arguments to wrapped function'''
24
25 self._showargs(*fargs, **kw)
26 ret=self.f(*fargs, **kw)
27 self._aftercall(ret)
28 return ret
29
30 def __repr__(self):
31 return self.f.func_name
32
33
34 class DecoTraceWithArgs(object):
35 '''decorator class with ARGUMENTS
36
37 This can be used for unbounded functions and methods. If this wraps a
38 class instance, then extract it and pass to the wrapped method as the
39 first arg.
40 '''
41
42 def __init__(self, *dec_args, **dec_kw):
43 '''The decorator arguments are passed here. Save them for runtime.'''
44 self.dec_args = dec_args
45 self.dec_kw = dec_kw
46
47 self.label = dec_kw.get('label', 'T')
48 self.fid = dec_kw.get('stream', stderr)
49
50 def _showargs(self, *fargs, **kw):
51
52 print >> self.fid, \
53 '%s: enter %s with args=%s, kw=%s' % (self.label, self.f.__name__, str(fargs), str(kw))
54 print >> self.fid, \
55 '%s: passing decorator args=%s, kw=%s' % (self.label, str(self.dec_args), str(self.dec_kw))
56
57 def _aftercall(self, status):
58 print >> self.fid, '%s: exit %s with status=%s' % (self.label, self.f.__name__, str(status))
59 def _showinstance(self, instance):
60 print >> self.fid, '%s: instance=%s' % (self.label, instance)
61
62 def __call__(self, f):
63 def wrapper(*fargs, **kw):
64 '''
65 Combine decorator arguments and function arguments and pass to wrapped
66 class instance-aware function/method.
67
68 Note: the first argument cannot be "self" because we get a parse error
69 "takes at least 1 argument" unless the instance is actually included in
70 the argument list, which is redundant. If this wraps a class instance,
71 the "self" will be the first argument.
72 '''
73
74 self._showargs(*fargs, **kw)
75
76 # merge decorator keywords into the kw argument list
77 kw.update(self.dec_kw)
78
79 # Does this wrap a class instance?
80 if fargs and getattr(fargs[0], '__class__', None):
81
82 # pull out the instance and combine function and
83 # decorator args
84 instance, fargs = fargs[0], fargs[1:]+self.dec_args
85 self._showinstance(instance)
86
87 # call the method
88 ret=f(instance, *fargs, **kw)
89 else:
90 # just send in the give args and kw
91 ret=f(*(fargs + self.dec_args), **kw)
92
93 self._aftercall(ret)
94 return ret
95
96 # Save wrapped function reference
97 self.f = f
98 wrapper.__name__ = f.__name__
99 wrapper.__dict__.update(f.__dict__)
100 wrapper.__doc__ = f.__doc__
101 return wrapper
102
103
104 @DecoTrace
105 def FirstBruce(*fargs, **kwargs):
106 'Simple function using simple decorator'
107 if fargs and fargs[0]:
108 print fargs[0]
109
110 @DecoTraceWithArgs(name="Second Bruce", standardline="Goodday, Bruce!")
111 def SecondBruce(*fargs, **kwargs):
112 'Simple function using decorator with arguments'
113 print '%s:' % kwargs.get('name', 'Unknown Bruce'),
114
115 if fargs and fargs[0]:
116 print fargs[0]
117 else:
118 print kwargs.get('standardline', None)
119
120 class Bruce(object):
121 'Simple class'
122
123 def __init__(self, id):
124 self.id = id
125
126 def __str__(self):
127 return self.id
128
129 def __repr__(self):
130 return 'Bruce'
131
132 @DecoTraceWithArgs(label="Trace a class", standardline="How are yer Bruce?", stream=stdout)
133 def talk(self, *fargs, **kwargs):
134 'Simple function using decorator with arguments'
135
136 print '%s:' % self,
137 if fargs and fargs[0]:
138 print fargs[0]
139 else:
140 print kwargs.get('standardline', None)
141
142 ThirdBruce = Bruce('Third Bruce')
143
144 SecondBruce()
145 FirstBruce("First Bruce: Oh, Hello Bruce!")
146 ThirdBruce.talk()
147 FirstBruce("First Bruce: Bit crook, Bruce.")
148 SecondBruce("Where's Bruce?")
149 FirstBruce("First Bruce: He's not here, Bruce")
150 ThirdBruce.talk("Blimey, s'hot in here, Bruce.")
151 FirstBruce("First Bruce: S'hot enough to boil a monkey's bum!")
152 SecondBruce("That's a strange expression, Bruce.")
153 FirstBruce("First Bruce: Well Bruce, I heard the Prime Minister use it. S'hot enough to boil a monkey's bum in 'ere, your Majesty,' he said and she smiled quietly to herself.")
154 ThirdBruce.talk("She's a good Sheila, Bruce and not at all stuck up.")
Unimplemented function replacement
Allows you to test unimplemented code in a development environment by specifying a default argument as an argument to the decorator (or you can leave it off to specify None to be returned.
1 # Annotation wrapper annotation method
2 def unimplemented(defaultval):
3 if(type(defaultval) == type(unimplemented)):
4 return lambda: None
5 else:
6 # Actual annotation
7 def unimp_wrapper(func):
8 # What we replace the function with
9 def wrapper(*arg):
10 return defaultval
11 return wrapper
12 return unimp_wrapper
Redirects stdout printing to python standard logging.
1 class LogPrinter:
2 """LogPrinter class which serves to emulates a file object and logs
3 whatever it gets sent to a Logger object at the INFO level."""
4 def __init__(self):
5 """Grabs the specific logger to use for logprinting."""
6 self.ilogger = logging.getLogger('logprinter')
7 il = self.ilogger
8 logging.basicConfig()
9 il.setLevel(logging.INFO)
10
11 def write(self, text):
12 """Logs written output to a specific logger"""
13 self.ilogger.info(text)
14
15 def logprintinfo(func):
16 """Wraps a method so that any calls made to print get logged instead"""
17 def pwrapper(*arg):
18 stdobak = sys.stdout
19 lpinstance = LogPrinter()
20 sys.stdout = lpinstance
21 try:
22 return func(*arg)
23 finally:
24 sys.stdout = stdobak
25 return pwrapper
This code needs a try/finally!
Access control
This example prevents users from getting access to places where they are not authorised to go
1 class LoginCheck:
2 """
3 This class checks whether a user
4 has logged in properly via
5 the global "check_function". If so,
6 the requested routine is called.
7 Otherwise, an alternative page is
8 displayed via the global "alt_function"
9 """
10 def __init__(self, f):
11 self._f = f
12
13 def __call__(self, *args):
14 Status = check_function()
15 if Status==1:
16 return self._f(*args)
17 else:
18 return alt_function()
19
20 def check_function():
21 return test
22
23 def alt_function():
24 return 'Sorry - this is the forced behaviour'
25
26 @LoginCheck
27 def display_members_page():
28 print 'This is the members page'
Example:
Events rising and handling
Please see the code and examples here: http://pypi.python.org/pypi/Decovent
Singleton
1 import functools
2
3 def singleton(cls):
4 """ Use class as singleton. """
5
6 cls.__new_original__ = cls.__new__
7
8 @functools.wraps(cls.__new__)
9 def singleton_new(cls, *args, **kw):
10 it = cls.__dict__.get('__it__')
11 if it is not None:
12 return it
13
14 cls.__it__ = it = cls.__new_original__(cls, *args, **kw)
15 it.__init_original__(*args, **kw)
16 return it
17
18 cls.__new__ = singleton_new
19 cls.__init_original__ = cls.__init__
20 cls.__init__ = object.__init__
21
22 return cls
23
24 #
25 # Sample use:
26 #
27
28 @singleton
29 class Foo:
30 def __new__(cls):
31 cls.x = 10
32 return object.__new__(cls)
33
34 def __init__(self):
35 assert self.x == 10
36 self.x = 15
37
38 assert Foo().x == 15
39 Foo().x = 20
40 assert Foo().x == 20
Asynchronous Call
1 from Queue import Queue
2 from threading import Thread
3
4 class asynchronous(object):
5 def __init__(self, func):
6 self.func = func
7
8 def threaded(*args, **kwargs):
9 self.queue.put(self.func(*args, **kwargs))
10
11 self.threaded = threaded
12
13 def __call__(self, *args, **kwargs):
14 return self.func(*args, **kwargs)
15
16 def start(self, *args, **kwargs):
17 self.queue = Queue()
18 thread = Thread(target = self.threaded, args = args, kwargs = kwargs);
19 thread.start();
20 return asynchronous.Result(self.queue, thread)
21
22 class NotYetDoneException(Exception):
23 def __init__(self, message):
24 self.message = message
25
26 class Result(object):
27 def __init__(self, queue, thread):
28 self.queue = queue
29 self.thread = thread
30
31 def is_done(self):
32 return not self.thread.is_alive()
33
34 def get_result(self):
35 if not self.is_done():
36 raise asynchronous.NotYetDoneException('the call has not yet completed its task')
37
38 if not hasattr(self, 'result'):
39 self.result = self.queue.get()
40
41 return self.result
42
43 if __name__ == '__main__':
44 # sample usage
45 import time
46
47 @asynchronous
48 def long_process(num):
49 time.sleep(10)
50 return num * num
51
52 result = long_process.start(12)
53
54 for i in range(20):
55 print i
56 time.sleep(1)
57
58 if result.is_done():
59 print "result {0}".format(result.get_result())
60
61
62 result2 = long_process.start(13)
63
64 try:
65 print "result2 {0}".format(result2.get_result())
66
67 except asynchronous.NotYetDoneException as ex:
68 print ex.message
Class method decorator using instance
When decorating a class method, the decorator receives an function not yet bound to an instance.
The decorator can't to do anything on the instance invocating it, unless it actually is a descriptor.
1 from functools import wraps
2
3 def decorate(f):
4 '''
5 Class method decorator specific to the instance.
6
7 It uses a descriptor to delay the definition of the
8 method wrapper.
9 '''
10 class descript(object):
11 def __init__(self, f):
12 self.f = f
13
14 def __get__(self, instance, klass):
15 if instance is None:
16 # Class method was requested
17 return self.make_unbound(klass)
18 return self.make_bound(instance)
19
20 def make_unbound(self, klass):
21 @wraps(self.f)
22 def wrapper(*args, **kwargs):
23 '''This documentation will vanish :)'''
24 raise TypeError(
25 'unbound method %s() must be called with %s instance '
26 'as first argument (got nothing instead)'
27 %
28 (self.f.__name__, klass.__name__)
29 )
30 return wrapper
31
32 def make_bound(self, instance):
33 @wraps(self.f)
34 def wrapper(*args, **kwargs):
35 '''This documentation will disapear :)'''
36 print "Called the decorated method %r of %r"%(self.f.__name__, instance)
37 return self.f(instance, *args, **kwargs)
38 # This instance does not need the descriptor anymore,
39 # let it find the wrapper directly next time:
40 setattr(instance, self.f.__name__, wrapper)
41 return wrapper
42
43 return descript(f)
This implementation replaces the descriptor by the actual decorated function a.s.a.p. to avoid overhead, but you could keep it to do even more (counting calls, etc...)