[Django]-Middleware for celery tasks

by

4👍

✅

This can be done using a base abstract task handler. For exceptions in particular there is a on_failure handler.

from celery import Task

class MyBaseTask(Task):
    abstract = True

    def on_failure(self, exc, task_id, args, kwargs, einfo):
        # Task failed. What do you want to do?
        print('Task raised an exception: {}'.format(exc)

@app.task(base=MyBaseTask)
def my_task():
    # Your task code

See the Celery docs for more info on the available handlers and the arguments which are passed to each one: http://docs.celeryproject.org/en/latest/userguide/tasks.html#abstract-classes

👤Mark Lavin

Leave a comment