Skip to content

Instantly share code, notes, and snippets.

@chadgh
Last active February 15, 2022 04:32
Show Gist options
  • Save chadgh/7093490 to your computer and use it in GitHub Desktop.
Save chadgh/7093490 to your computer and use it in GitHub Desktop.
python decorators for; debugging, threading
def timeit(function):
'''Decorator used for debugging. Prints the call and how long it took.'''
def timed(*args, **kwargs):
ts = time.time()
result = function(*args, **kwargs)
te = time.time()
print("{0} ({1}, {2}) {3:.2} sec"
.format(function.__name__, args, kwargs, te - ts))
return result
return timed
def debug(function):
'''Decorator that places a break point right before calling the function.'''
def wrapped(*args, **kwargs):
import pdb; pdb.set_trace() # XXX BREAKPOINT
return function(*args, **kwargs)
return wrapped
import functools
def decorator(decorator_func):
'''Decorator to decorate decorators. Make sure to call the
func with this decorator attached.
'''
decorator_expected_arg_count = decorator_func.__code__.co_argcount - 1
if decorator_expected_arg_count:
def decorator_maker(*decorator_args):
assert len(decorator_args) == decorator_expected_arg_count,\
"%s expected %d args" % (decorator_func.__name__,
decorator_expected_arg_count)
def _decorator(func):
assert callable(func), \
"Decorator not given a function. Did you forget to \
give %r arguments?" % (decorator_func.__name__)
def decorated_func(*args, **kwargs):
full_args = decorator_args + (func,) + args
return decorator_func(*full_args, **kwargs)
decorated_func.__name__ = func.__name__
decorated_func.__doc__ = func.__doc__
return decorated_func
return _decorator
return decorator_maker
else:
def _decorator(func):
def decorated_func(*args, **kwargs):
return decorator_func(func, *args, **kwargs)
decorated_func.__name__ = func.__name__
decorated_func.__doc__ = func.__doc__
return decorated_func
return _decorator
def notify(email=None,
email_from='[email protected]',
email_subject='Notification',
smtp_host='localhost',
log=None,
*args, **kwargs):
'''Decorator to send email notification or log a message to a log file
or both with the results of running the function.
Attributes:
email - List of email addresses to send results to
email_from - Email address to use for from address
email_subject - Subject to use for the emails
smtp_host - SMTP host to connect to for sending emails
log - Log file location to log notifications to
Example:
@notify(email=['[email protected]', '[email protected]'])
def do_something():
return 'this is what was done'
Executing the do_something function will then email the email addresses
with the results of the function.
'''
DEFAULT_LOG_FILE = '/tmp/default_log.log'
def wrap(func):
name = func.__name__
@functools.wraps(func)
def wrapper(*args, **kwargs):
logfile = DEFAULT_LOG_FILE
rtn = func(*args, **kwargs)
message = "notify: {0}({2}, {3}) -> {1}\n".format(name,
rtn,
str(args),
str(kwargs))
if log is not None:
logfile = log
with open(logfile, 'a') as f:
f.write(message)
if email is not None and len(email) > 0:
import smtplib
from email.mime.multipart import MIMEMultipart
msg = MIMEMultipart()
msg['Subject'] = email_subject
msg['From'] = email_from
msg['To'] = ', '.join(email)
try:
s = smtplib.SMTP(smtp_host)
s.sendmail(email_from, email, msg.as_string())
s.quit()
except Exception:
with open(logfile, 'a') as f:
f.write('could not send notification via email\n')
return rtn
return wrapper
return wrap
def memorize(func):
'''Decorator. Memorizes the results of calling a function with
specific args and kwargs. If the function is called again with
those same args and kwargs the previous result is returned. The
function is no actually executed again.
'''
cache = func.cache = {}
@functools.wraps(func)
def memorizer(*args, **kwargs):
key = str(args) + str(kwargs)
if key not in cache:
cache[key] = func(*args, **kwargs)
return cache[key]
return memorizer
@decorator
def deprecated(func, *args, **kwargs):
'''Decorator which can be used to mark functions as deprecated.
It will result in a warning being emitted the the function is
called.
'''
import warnings
warnings.warn('Call to deprecated function{}.'.format(func.__name__),
category=DeprecationWarning)
return func(*args, **kwargs)
def threadify(func, daemon=False):
'''Decorator adapted from http://stackoverflow.com/a/14331755/18992
(thanks bj0)
'''
import queue
import threading
def wrapped_f(q, *args, **kwargs):
rtn = func(*args, **kwargs)
q.put(rtn)
@functools.wraps(func)
def wrap(*args, **kwargs):
q = queue.Queue()
t = threading.Thread(target=wrapped_f, args=(q,) + args, kwargs=kwargs)
t.daemon = daemon
t.start()
t.result = q
return t
return wrap
@gonzaloamadio
Copy link

I tried to use @deprecated, but it gives me an error in decorator function. In this line

decorator_expected_arg_count = decorator_func.func_code.co_argcount - 1
AttributeError: 'function' object has no attribute 'func_code'

Also, Is there a way to enable or disable decorators if we are in dev mode , or in DEBUG?
And if you have some examples of how you used this decorators, it will help to.

Cheers!

@chadgh
Copy link
Author

chadgh commented Nov 4, 2019

@gonzaloamadio

This gist currently only works with Python 2.7. I'll have to update it to work with Python 3 at some point.

As for enabling and disabling, you could modify these decorators to check an environment variable to know whether or not they should run.

@gonzaloamadio
Copy link

I executed 2to3 converter. This lines need to be changed

  • decorator_expected_arg_count = decorator_func.func_code.co_argcount - 1
  • decorator_expected_arg_count = decorator_func.code.co_argcount - 1
  • import Queue
  • import queue
  •    q = Queue.Queue()
    
  •    q = queue.Queue()
    

@gonzaloamadio
Copy link

Here you have my updated file if u want to update it : https://gist.github.com/gonzaloamadio/22debbaaa8695f89ad6a51b17b32f6d1/revisions

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment