Skip to content

Instantly share code, notes, and snippets.

What would you like to do?
Log execution time for every Celery task, working on Celery 3.1.x.
import logging
import time
from celery import shared_task as orig_shared_task
def timeit(func):
"""A decorator used to log the function execution time."""
logger = logging.getLogger('tasks')
# Use functools.wraps to ensure function name is not changed
def wrap(*args, **kwargs):
start = time.time()
result = func(*args, **kwargs)
dur = time.time() - start
msg = {
'task_name': func.__module__ + '.' + func.__name__,
'duration': dur,
# Use 'task_args' instead of 'args' because 'args' conflicts with
# attribute of LogRecord
'task_args': args,
'task_kwargs': kwargs
}'%s %.2fs', func.__name__, dur, extra=msg)
return result
return wrap
def shared_task(*args, **kwargs):
"""Override Celery's default shared_task decorator to log every task call.
if len(args) == 1 and callable(args[0]):
func = args[0]
return orig_shared_task(**kwargs)(timeit(func))
def decorator(func):
return orig_shared_task(*args, **kwargs)(timeit(func))
return decorator

This comment has been minimized.

Copy link

@oca159 oca159 commented Dec 16, 2016

Could you please give me an example about how can I use this script?

I am trying to log the last execution datetime.
My idea is read and write in a file, for example a python script.
I need this information for some tasks that I am using to retrieve data base on a date.
Thanks a lot.

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