Can I add an on_failure callback to a celery task created using a task decorator?

Question

I am sure that this can only be done if I create my own class of tasks, but I would like to know if anyone else has found a way to do this.

Answer 1

The answer is yes, and it is quite easy.

http://docs.celeryproject.org/en/latest/userguide/tasks.html#abstract-classes

Answer 2

To expand on the accepted answer, here is a complete solution (works for Celery 4+ ):

 import celery from celery.task import task class MyBaseClassForTask(celery.Task): def on_failure(self, exc, task_id, args, kwargs, einfo): # exc (Exception) - The exception raised by the task. # args (Tuple) - Original arguments for the task that failed. # kwargs (Dict) - Original keyword arguments for the task that failed. print('{0!r} failed: {1!r}'.format(task_id, exc)) @task(name="foo:my_task", base=MyBaseClassForTask) def add(x, y): raise KeyError() 

Resources

• http://docs.celeryproject.org/en/latest/userguide/tasks.html#task-inheritance
• http://docs.celeryproject.org/en/latest/reference/celery.app.task.html#celery.app.task.Task.on_failure