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 {}'.format(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)

 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

 1 class Example(object):
 2  @apply # doesn't exist in Python 3
 3  def myattr():
 4  doc = '''This is the doc string.'''
 5 
 6  def fget(self):
 7  return self._half * 2
 8 
 9  def fset(self, value):
 10  self._half = value / 2
 11 
 12  def fdel(self):
 13  del self._half
 14 
 15  return property(**locals())
 16  #myattr = myattr() # works in Python 2 and 3

 1 try:
 2  # Python 2
 3  import __builtin__ as builtins
 4 except ImportError:
 5  # Python 3
 6  import builtins
 7 
 8 def property(function):
 9  keys = 'fget', 'fset', 'fdel'
 10  func_locals = {'doc':function.__doc__}
 11  def probe_func(frame, event, arg):
 12  if event == 'return':
 13  locals = frame.f_locals
 14  func_locals.update(dict((k, locals.get(k)) for k in keys))
 15  sys.settrace(None)
 16  return probe_func
 17  sys.settrace(probe_func)
 18  function()
 19  return builtins.property(**func_locals)
 20 
 21 #====== Example =======================================================
 22 
 23 from math import radians, degrees, pi
 24 
 25 class Angle(object):
 26  def __init__(self, rad):
 27  self._rad = rad
 28 
 29  @property
 30  def rad():
 31  '''The angle in radians'''
 32  def fget(self):
 33  return self._rad
 34  def fset(self, angle):
 35  if isinstance(angle, Angle):
 36  angle = angle.rad
 37  self._rad = float(angle)
 38 
 39  @property
 40  def deg():
 41  '''The angle in degrees'''
 42  def fget(self):
 43  return degrees(self._rad)
 44  def fset(self, angle):
 45  if isinstance(angle, Angle):
 46  angle = angle.deg
 47  self._rad = radians(angle)

 1 import collections
 2 import functools
 3 
 4 class memoized(object):
 5  '''Decorator. Caches a function's return value each time it is called.
 6  If called later with the same arguments, the cached value is returned
 7  (not reevaluated).
 8  '''
 9  def __init__(self, func):
 10  self.func = func
 11  self.cache = {}
 12  def __call__(self, *args):
 13  if not isinstance(args, collections.Hashable):
 14  # uncacheable. a list, for instance.
 15  # better to not cache than blow up.
 16  return self.func(*args)
 17  if args in self.cache:
 18  return self.cache[args]
 19  else:
 20  value = self.func(*args)
 21  self.cache[args] = value
 22  return value
 23  def __repr__(self):
 24  '''Return the function's docstring.'''
 25  return self.func.__doc__
 26  def __get__(self, obj, objtype):
 27  '''Support instance methods.'''
 28  return functools.partial(self.__call__, obj)
 29 
 30 @memoized
 31 def fibonacci(n):
 32  "Return the nth fibonacci number."
 33  if n in (0, 1):
 34  return n
 35  return fibonacci(n-1) + fibonacci(n-2)
 36 
 37 print fibonacci(12)

 1 # note that this decorator ignores **kwargs
 2 def memoize(obj):
 3  cache = obj.cache = {}
 4 
 5  @functools.wraps(obj)
 6  def memoizer(*args, **kwargs):
 7  if args not in cache:
 8  cache[args] = obj(*args, **kwargs)
 9  return cache[args]
 10  return memoizer

 1 def memoize(obj):
 2  cache = obj.cache = {}
 3 
 4  @functools.wraps(obj)
 5  def memoizer(*args, **kwargs):
 6  key = str(args) + str(kwargs)
 7  if key not in cache:
 8  cache[key] = obj(*args, **kwargs)
 9  return cache[key]
 10  return memoizer

 1 class memoize(dict):
 2  def __init__(self, func):
 3  self.func = func
 4 
 5  def __call__(self, *args):
 6  return self[args]
 7 
 8  def __missing__(self, key):
 9  result = self[key] = self.func(*key)
 10  return result
 11 
 12 #
 13 # Sample use
 14 #
 15 
 16 >>> @memoize
 17 ... def foo(a, b):
 18 ... return a * b
 19 >>> foo(2, 4)
 20 8
 21 >>> foo
 22 {(2, 4): 8}
 23 >>> foo('hi', 3)
 24 'hihihi'
 25 >>> foo
 26 {(2, 4): 8, ('hi', 3): 'hihihi'}

 1 import pickle
 2 import collections
 3 import functools
 4 import inspect
 5 import os.path
 6 import re
 7 import unicodedata
 8 
 9 class Memorize(object):
 10  '''
 11  A function decorated with @Memorize caches its return
 12  value every time it is called. If the function is called
 13  later with the same arguments, the cached value is
 14  returned (the function is not reevaluated). The cache is
 15  stored as a .cache file in the current directory for reuse
 16  in future executions. If the Python file containing the
 17  decorated function has been updated since the last run,
 18  the current cache is deleted and a new cache is created
 19  (in case the behavior of the function has changed).
 20  '''
 21  def __init__(self, func):
 22  self.func = func
 23  self.set_parent_file() # Sets self.parent_filepath and self.parent_filename
 24  self.__name__ = self.func.__name__
 25  self.set_cache_filename()
 26  if self.cache_exists():
 27  self.read_cache() # Sets self.timestamp and self.cache
 28  if not self.is_safe_cache():
 29  self.cache = {}
 30  else:
 31  self.cache = {}
 32 
 33  def __call__(self, *args):
 34  if not isinstance(args, collections.Hashable):
 35  return self.func(*args)
 36  if args in self.cache:
 37  return self.cache[args]
 38  else:
 39  value = self.func(*args)
 40  self.cache[args] = value
 41  self.save_cache()
 42  return value
 43 
 44  def set_parent_file(self):
 45  """
 46  Sets self.parent_file to the absolute path of the
 47  file containing the memoized function.
 48  """
 49  rel_parent_file = inspect.stack()[-1].filename
 50  self.parent_filepath = os.path.abspath(rel_parent_file)
 51  self.parent_filename = _filename_from_path(rel_parent_file)
 52 
 53  def set_cache_filename(self):
 54  """
 55  Sets self.cache_filename to an os-compliant
 56  version of "file_function.cache"
 57  """
 58  filename = _slugify(self.parent_filename.replace('.py', ''))
 59  funcname = _slugify(self.__name__)
 60  self.cache_filename = filename+'_'+funcname+'.cache'
 61 
 62  def get_last_update(self):
 63  """
 64  Returns the time that the parent file was last
 65  updated.
 66  """
 67  last_update = os.path.getmtime(self.parent_filepath)
 68  return last_update
 69 
 70  def is_safe_cache(self):
 71  """
 72  Returns True if the file containing the memoized
 73  function has not been updated since the cache was
 74  last saved.
 75  """
 76  if self.get_last_update() > self.timestamp:
 77  return False
 78  return True
 79 
 80  def read_cache(self):
 81  """
 82  Read a pickled dictionary into self.timestamp and
 83  self.cache. See self.save_cache.
 84  """
 85  with open(self.cache_filename, 'rb') as f:
 86  data = pickle.loads(f.read())
 87  self.timestamp = data['timestamp']
 88  self.cache = data['cache']
 89 
 90  def save_cache(self):
 91  """
 92  Pickle the file's timestamp and the function's cache
 93  in a dictionary object.
 94  """
 95  with open(self.cache_filename, 'wb+') as f:
 96  out = dict()
 97  out['timestamp'] = self.get_last_update()
 98  out['cache'] = self.cache
 99  f.write(pickle.dumps(out))
 100 
 101  def cache_exists(self):
 102  '''
 103  Returns True if a matching cache exists in the current directory.
 104  '''
 105  if os.path.isfile(self.cache_filename):
 106  return True
 107  return False
 108 
 109  def __repr__(self):
 110  """ Return the function's docstring. """
 111  return self.func.__doc__
 112 
 113  def __get__(self, obj, objtype):
 114  """ Support instance methods. """
 115  return functools.partial(self.__call__, obj)
 116 
 117 def _slugify(value):
 118  """
 119  Normalizes string, converts to lowercase, removes
 120  non-alpha characters, and converts spaces to
 121  hyphens. From
 122  http://stackoverflow.com/questions/295135/turn-a-string-into-a-valid-filename-in-python
 123  """
 124  value = unicodedata.normalize('NFKD', value).encode('ascii', 'ignore')
 125  value = re.sub(r'[^\w\s-]', '', value.decode('utf-8', 'ignore'))
 126  value = value.strip().lower()
 127  value = re.sub(r'[-\s]+', '-', value)
 128  return value
 129 
 130 def _filename_from_path(filepath):
 131  return filepath.split('/')[-1]

 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 create 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

 1 import time
 2 import math
 3 
 4 # Retry decorator with exponential backoff
 5 def retry(tries, delay=3, backoff=2):
 6  '''Retries a function or method until it returns True.
 7 
 8  delay sets the initial delay in seconds, and backoff sets the factor by which
 9  the delay should lengthen after each failure. backoff must be greater than 1,
 10  or else it isn't really a backoff. tries must be at least 0, and delay
 11  greater than 0.'''
 12 
 13  if backoff <= 1:
 14  raise ValueError("backoff must be greater than 1")
 15 
 16  tries = math.floor(tries)
 17  if tries < 0:
 18  raise ValueError("tries must be 0 or greater")
 19 
 20  if delay <= 0:
 21  raise ValueError("delay must be greater than 0")
 22 
 23  def deco_retry(f):
 24  def f_retry(*args, **kwargs):
 25  mtries, mdelay = tries, delay # make mutable
 26 
 27  rv = f(*args, **kwargs) # first attempt
 28  while mtries > 0:
 29  if rv is True: # Done on success
 30  return True
 31 
 32  mtries -= 1 # consume an attempt
 33  time.sleep(mdelay) # wait...
 34  mdelay *= backoff # make future wait longer
 35 
 36  rv = f(*args, **kwargs) # Try again
 37 
 38  return False # Ran out of tries :-(
 39 
 40  return f_retry # true decorator -> decorated function
 41  return deco_retry # @retry(arg[, ...]) -> true decorator

 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 
 12  def __call__(self, *a):
 13  args = self.args + a
 14  if len(args) < self.func.func_code.co_argcount:
 15  return curried(self.func, *args)
 16  else:
 17  return self.func(*args)
 18 
 19 
 20 @curried
 21 def add(a, b):
 22  return a + b
 23 
 24 add1 = add(1)
 25 
 26 print add1(2)

 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

 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 is 5

 1 @decorator
 2 def my_property(getter, *, setter=None, deleter=None, doc=None):
 3  return property(getter, setter, deleter, doc)

 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))

 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"

 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])

 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

 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 {}.".format(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

 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 {}.".format(func.__name__),
 14  category=DeprecationWarning,
 15  filename=func.func_code.co_filename,
 16  lineno=func.func_code.co_firstlineno + 1
 17  )
 18  return func(*args, **kwargs)
 19  return new_func
 20 
 21 
 22 ## Usage examples ##
 23 @deprecated
 24 def my_func():
 25  pass
 26 
 27 @other_decorators_must_be_upper
 28 @deprecated
 29 def my_func():
 30  pass

 1 import warnings
 2 
 3 def ignore_deprecation_warnings(func):
 4  '''This is a decorator which can be used to ignore deprecation warnings
 5  occurring in a function.'''
 6  def new_func(*args, **kwargs):
 7  with warnings.catch_warnings():
 8  warnings.filterwarnings("ignore", category=DeprecationWarning)
 9  return func(*args, **kwargs)
 10  new_func.__name__ = func.__name__
 11  new_func.__doc__ = func.__doc__
 12  new_func.__dict__.update(func.__dict__)
 13  return new_func
 14 
 15 # === Examples of use ===
 16 
 17 @ignore_deprecation_warnings
 18 def some_function_raising_deprecation_warning():
 19  warnings.warn("This is a deprecationg warning.",
 20  category=DeprecationWarning)
 21 
 22 class SomeClass:
 23  @ignore_deprecation_warnings
 24  def some_method_raising_deprecation_warning():
 25  warnings.warn("This is a deprecationg warning.",
 26  category=DeprecationWarning)

 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"

 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 
 6  def echo_func(*args,**kwargs):
 7  print fname, ":", ', '.join(
 8  '%s=%r' % entry
 9  for entry in zip(argnames,args) + kwargs.items())
 10  return func(*args, **kwargs)
 11 
 12  return echo_func
 13 
 14 @dump_args
 15 def f1(a,b,c):
 16  print a + b + c
 17 
 18 f1(1, 2, 3)

 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()

 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 "{}({}): {}".format( bname,
 19  lineno,
 20  linecache.getline(filename, lineno)),
 21  return localtrace
 22 
 23  def _f(*args, **kwds):
 24  sys.settrace(globaltrace)
 25  result = f(*args, **kwds)
 26  sys.settrace(None)
 27  return result
 28 
 29  return _f

 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

 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  #!python
 6 -- MEDIUM: Print warning message to stderr. (Default)
 7  2 -- STRONG: Raise TypeError with message.
 8 If 'debug' is not passed to the decorator, the default level is used.
 9 
 10 Example usage:
 11  >>> NONE, MEDIUM, STRONG = 0, 1, 2
 12  >>>
 13  >>> @accepts(int, int, int)
 14  ... @returns(float)
 15  ... def average(x, y, z):
 16  ... return (x + y + z) / 2
 17  ...
 18  >>> average(5.5, 10, 15.0)
 19  TypeWarning: 'average' method accepts (int, int, int), but was given
 20  (float, int, float)
 21  15.25
 22  >>> average(5, 10, 15)
 23  TypeWarning: 'average' method returns (float), but result is (int)
 24  15
 25 
 26 Needed to cast params as floats in function def (or simply divide by 2.0).
 27 
 28  >>> TYPE_CHECK = STRONG
 29  >>> @accepts(int, debug=TYPE_CHECK)
 30  ... @returns(int, debug=TYPE_CHECK)
 31  ... def fib(n):
 32  ... if n in (0, 1): return n
 33  ... return fib(n-1) + fib(n-2)
 34  ...
 35  >>> fib(5.3)
 36  Traceback (most recent call last):
 37  ...
 38  TypeError: 'fib' method accepts (int), but was given (float)
 39 
 40 '''
 41 import sys
 42 
 43 def accepts(*types, **kw):
 44  '''Function decorator. Checks decorated function's arguments are
 45  of the expected types.
 46 
 47  Parameters:
 48  types -- The expected types of the inputs to the decorated function.
 49  Must specify type for each parameter.
 50  kw -- Optional specification of 'debug' level (this is the only valid
 51  keyword argument, no other should be given).
 52  debug = ( 0 | 1 | 2 )
 53 
 54  '''
 55  if not kw:
 56  # default level: MEDIUM
 57  debug = 1
 58  else:
 59  debug = kw['debug']
 60  try:
 61  def decorator(f):
 62  def newf(*args):
 63  if debug is 0:
 64  return f(*args)
 65  assert len(args) == len(types)
 66  argtypes = tuple(map(type, args))
 67  if argtypes != types:
 68  msg = info(f.__name__, types, argtypes, 0)
 69  if debug is 1:
 70  print >> sys.stderr, 'TypeWarning: ', msg
 71  elif debug is 2:
 72  raise TypeError, msg
 73  return f(*args)
 74  newf.__name__ = f.__name__
 75  return newf
 76  return decorator
 77  except KeyError, key:
 78  raise KeyError, key + "is not a valid keyword argument"
 79  except TypeError, msg:
 80  raise TypeError, msg
 81 
 82 
 83 def returns(ret_type, **kw):
 84  '''Function decorator. Checks decorated function's return value
 85  is of the expected type.
 86 
 87  Parameters:
 88  ret_type -- The expected type of the decorated function's return value.
 89  Must specify type for each parameter.
 90  kw -- Optional specification of 'debug' level (this is the only valid
 91  keyword argument, no other should be given).
 92  debug=(0 | 1 | 2)
 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 is 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 is 1:
 109  print >> sys.stderr, 'TypeWarning: ', msg
 110  elif debug is 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 = "'{}' method ".format( fname )\
 126  + ("accepts", "returns")[flag] + " ({}), but ".format(expected)\
 127  + ("was given", "result is")[flag] + " ({})".format(actual)
 128  return msg

 1 class CGImethod(object):
 2  def __init__(self, title):
 3  self.title = title
 4 
 5  def __call__(self, fn):
 6  def wrapped_fn(*args):
 7  print "Content-Type: text/html\n\n"
 8  print "<HTML>"
 9  print "<HEAD><TITLE>{}</TITLE></HEAD>".format(self.title)
 10  print "<BODY>"
 11  try:
 12  fn(*args)
 13  except Exception, e:
 14  print
 15  print e
 16  print
 17  print "</BODY></HTML>"
 18 
 19  return wrapped_fn
 20 
 21 @CGImethod("Hello with Decorator")
 22 def say_hello():
 23  print '<h1>Hello from CGI-Land</h1>'

 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 
 26  @event_handler(gstate)
 27  def event_b(self): pass
 28 
 29  @event_handler(gstate)
 30  def event_c(self, val): pass
 31 
 32  @event_handler(tstate)
 33  def toggle(self): pass
 34 
 35 
 36  # define methods to handle events.
 37  def _event_a_hdlr1(self):
 38  print "State 1, event A"
 39  self.a_count += 1
 40  def _event_b_hdlr1(self):
 41  print "State 1, event B"
 42  self.b_count += 1
 43  def _event_c_hdlr1(self, val):
 44  print "State 1, event C"
 45  self.c_count += 3*val
 46 
 47  def _event_a_hdlr2(self):
 48  print "State 2, event A"
 49  self.a_count += 10
 50  # here we brute force the tstate to on, leave & enter functions called if state changes.
 51  # turtle is object's state variable for tstate, comes from constructor argument
 52  self.turtle.set_state(self, self._t_on)
 53  def _event_b_hdlr2(self):
 54  print "State 2, event B"
 55  self.b_count += 10
 56  def _event_c_hdlr2(self, val):
 57  print "State 2, event C"
 58  self.c_count += 2*val
 59 
 60  def _event_a_hdlr3(self):
 61  self.a_count += 100
 62  print "State 3, event A"
 63  def _event_b_hdlr3(self):
 64  print "State 3, event B"
 65  self.b_count += 100
 66  # we decide here we want to go to state 2, overrrides spec in state table below.
 67  # transition to next_state is made after the method exits.
 68  self.gstate.next_state = self._state2
 69  def _event_c_hdlr3(self, val):
 70  print "State 3, event C"
 71  self.c_count += 5*val
 72 
 73  # Associate the handlers with a state. The first argument is a list of methods.
 74  # One method for each event_handler decorated function of gstate. Order of methods
 75  # in the list correspond to order in which the Event Handlers were declared.
 76  # Second arg is the name of the state. Third argument is to be come a list of the
 77  # next states.
 78  # The first state created becomes the initial state.
 79  _state1 = gstate.state("One", (_event_a_hdlr1, _event_b_hdlr1, _event_c_hdlr1),
 80  ("Two", "Three", None))
 81  _state2 = gstate.state("Two", (_event_a_hdlr2, _event_b_hdlr2, _event_c_hdlr2),
 82  ("Three", None, "One"))
 83  _state3 = gstate.state("Three",(_event_a_hdlr3, _event_b_hdlr3, _event_c_hdlr3),
 84  (None, "One", "Two"))
 85 
 86 
 87  # Declare a function that will be called when entering a new gstate.
 88  # Can also declare a leave function using @on_leave_function(gstate)
 89  @on_enter_function(gstate)
 90  def _enter_gstate(self):
 91  print "entering state ", self.gstate.name() , "of ", self.mname
 92  @on_leave_function(tstate)
 93  def _leave_tstate(self):
 94  print "leaving state ", self.turtle.name() , "of ", self.mname
 95 
 96 
 97  def _toggle_on(self):
 98  print "Toggle On"
 99 
 100  def _toggle_off(self):
 101  print "Toggle Off"
 102 
 103  _t_off = tstate.state("Off", [_toggle_on],
 104  ["On"])
 105  _t_on = tstate.state("On", [_toggle_off],
 106  ["Off"])
 107 
 108 
 109 def main():
 110  big_machine = MyMachine("big")
 111  lil_machine = MyMachine("lil")
 112 
 113  big_machine.event_a()
 114  lil_machine.event_a()
 115  big_machine.event_a()
 116  lil_machine.event_a()
 117  big_machine.event_b()
 118  lil_machine.event_b()
 119  big_machine.event_c(4)
 120  lil_machine.event_c(2)
 121  big_machine.event_c(1)
 122  lil_machine.event_c(3)
 123  big_machine.event_b()
 124  lil_machine.event_b()
 125  big_machine.event_a()
 126  lil_machine.event_a()
 127  big_machine.event_a()
 128 
 129  big_machine.toggle()
 130  big_machine.toggle()
 131  big_machine.toggle()
 132 
 133  lil_machine.event_a()
 134  big_machine.event_b()
 135  lil_machine.event_b()
 136  big_machine.event_c(3)
 137  big_machine.event_a()
 138  lil_machine.event_c(2)
 139  lil_machine.event_a()
 140  big_machine.event_b()
 141  lil_machine.event_b()
 142  big_machine.event_c(7)
 143  lil_machine.event_c(1)
 144 
 145  print "Event A count ", big_machine.a_count
 146  print "Event B count ", big_machine.b_count
 147  print "Event C count ", big_machine.c_count
 148  print "LilMachine C count ", lil_machine.c_count
 149 
 150 main()

 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: {}'.format(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