Skip to content

Instantly share code, notes, and snippets.

@dzhou
Created May 8, 2012 03:19
Show Gist options
  • Save dzhou/2632285 to your computer and use it in GitHub Desktop.
Save dzhou/2632285 to your computer and use it in GitHub Desktop.
useful python decorator utils
#!/usr/bin/env python
#
import os
import time
import thread
import warnings
import signal
import traceback
from threading import Timer
__all__=["none", "attempt_retries", "ignore_exception",
"asynchronous", "timer", "abstract", "synchronized",
"count_calls", "memoized", "deprecated", "time_out",
"attempt_retries_details"]
def none(func):
"""
Null decorator that does absolutely nothing
Useful for overriding other decorators in mock testing
"""
return func
def attempt_retries(func, retries=3, delay=0.3, IgnoreException=Exception):
"""
Decorator for ignoring certain exception for certain number times and
retrying with certain delay
e.g. func = @retry(connect_imap, 10, 5, SocketError)
e.g.2. retry(connect_imap)(username, password)
@returns the original function with an external retry wrapper
"""
def dec(*args, **kwargs):
for i in range(retries-1):
try:
return func(*args, **kwargs)
except IgnoreException:
time.sleep(delay)
out = func(*args, **kwargs)
return dec
# capture err details
def attempt_retries_detailed(func, retries=3,
delay=0.3, IgnoreException=Exception):
"""
Decorator for ignoring certain exception for certain number
times and retrying with certain delay
e.g. func = @retry(connect_imap, 10, 5, SocketError) # Tries ten times
e.g.2. retry(connect_imap)(username, password)
@returns the original function with an external retry wrapper
"""
def dec(*args, **kwargs):
for i in range(retries-1):
try:
return func(*args, **kwargs)
except IgnoreException:
time.sleep(delay)
try:
out = func(*args, **kwargs)
except Exception as e:
err = traceback.format_exc()
err += "%s\n" % e.__class__
if args:
err += "\tArgs: (%s)\n" % str(args[:10])
if kwargs:
err += "\tKwargs: (%s)" % str(kwargs.items()[:10])
raise Exception(err)
return dec
def ignore_exception(func):
"""
Decorator for supressing all exceptions
"""
def dec(*args, **kwargs):
try:
return func(*args, **kwargs)
except:
pass
return dec
def asynchronous(func):
"""
Run function in another thread.
Caller with no longer be blocked by this function, but also will not
be able to catch exception or get results from function.
"""
def dec(*args, **kwds):
thread.start_new_thread(func, args, kwds)
return dec
def abstract(func):
"""
Mark function as abstract.
Function will raise exception unless it is overridden in subclass
"""
def dec(*args, **kwds):
raise Exception("Abstract function %s() must be overridden" % func.__name__)
return dec
def synchronized(lock):
""" Synchronization decorator """
def wrap(f):
def dec(*args, **kw):
lock.acquire()
try:
return f(*args, **kw)
finally:
lock.release()
return dec
return wrap
def deprecated(func):
"""
This is a decorator which can be used to mark functions
as deprecated. It will result in a warning being emmitted
when the function is used.
"""
def newFunc(*args, **kwargs):
warnings.warn("Call to deprecated function %s." % func.__name__,
category=DeprecationWarning)
return func(*args, **kwargs)
newFunc.__name__ = func.__name__
newFunc.__doc__ = func.__doc__
newFunc.__dict__.update(func.__dict__)
return newFunc
class memoized(object):
"""
Decorator that caches a function's return value each time it is called.
If called later with the same arguments, the cached value is returned, and
not re-evaluated.
"""
def __init__(self, func):
self.func = func
self.cache = {}
def __call__(self, *args):
try:
return self.cache[args]
except KeyError:
self.cache[args] = value = self.func(*args)
return value
except TypeError:
# uncachable -- for instance, passing a list as an argument.
# Better to not cache than to blow up entirely.
return self.func(*args)
def __repr__(self):
"""Return the function's docstring."""
return self.func.__doc__
class TimedOutException(Exception):
def __init__(self, value="timed out"):
self.value = value
def __str__(self):
return repr(self.value)
def time_out(timeout):
def decorate(func):
def timeout_handler():
thread.interrupt_main()
def new_f(*args, **kwargs):
t1 = time.time()
timer = Timer(timeout, timeout_handler)
timer.start()
try:
return func(*args, **kwargs)
except:
if time.time() - t1 >= timeout:
raise TimedOutException("%s timed out after %ss" % (func.__name__, timeout))
else:
raise Exception(traceback.format_exc())
finally:
timer.cancel()
timer = None
new_f.__name__ = func.__name__
new_f.__doc__ = func.__doc__
new_f.__dict__.update(func.__dict__)
return new_f
return decorate
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment