def assertThread(*threads):
"""Decorator that asserts the wrapped function is only
run in a given thread
"""
# If no thread list was supplied, assume the wrapped
# function should be run in the current thread.
if not threads:
threads = (threading.currentThread(), )
def decorator(func):
def wrapper(*args, **kw):
currentThread = threading.currentThread()
name = currentThread.getName()
assert currentThread in threads or name in threads, \
"%s erroniously called in %s thread " \
context" % (func.__name__, name)
return func(*args, **kw)
if __debug__:
return wrapper
else:
return func
return decorator
You can restrict execution to one or more threads, each specified by either the thread object or thread name.
Note the trick at the end to make the decorator effectively a no-op in production. Using this decorator around your functions and methods helps you spot logic errors during development without impacting the performance of your production code. Of course, if you are of the school that assertions should never be disabled, feel free to replace the final
if __debug__:
/else
block with an unconditional return of wrapper
.
No comments:
Post a Comment