Last active
December 16, 2016 17:45
-
-
Save eliangcs/a7716b6e361162f451cd to your computer and use it in GitHub Desktop.
Log execution time for every Celery task, working on Celery 3.1.x.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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 | |
# http://stackoverflow.com/questions/13492603/celery-task-with-multiple-decorators-not-auto-registering-task-name | |
@functools.wraps(func) | |
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 | |
} | |
logger.info('%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 |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
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.