Reputation: 1971
I use celery
in my application to run periodic tasks. Let's see simple example below
from myqueue import Queue
@perodic_task(run_every=timedelta(minutes=1))
def process_queue():
queue = Queue()
uid, questions = queue.pop()
if uid is None:
return
job = group(do_stuff(q) for q in questions)
job.apply_async()
def do_stuff(question):
try:
...
except:
...
raise
As you can see in the example above, i use celery
to run async task, but (since it's a queue) i need to do queue.fail(uid)
in case of exception in do_stuff
or queue.ack(uid)
otherwise. In this situation it would be very clear and usefull to have some callback from my task in both cases - on_failure
and on_success
.
I saw some documentation, but never seen practices of using callbacks with apply_async
. Is it possible to do that?
Upvotes: 17
Views: 39696
Reputation: 1140
Subclass the Task class and overload the on_success and on_failure functions:
from celery import Task
class CallbackTask(Task):
def on_success(self, retval, task_id, args, kwargs):
'''
retval – The return value of the task.
task_id – Unique id of the executed task.
args – Original arguments for the executed task.
kwargs – Original keyword arguments for the executed task.
'''
pass
def on_failure(self, exc, task_id, args, kwargs, einfo):
'''
exc – The exception raised by the task.
task_id – Unique id of the failed task.
args – Original arguments for the task that failed.
kwargs – Original keyword arguments for the task that failed.
'''
pass
Use:
@celery.task(base=CallbackTask) # this does the trick
def add(x, y):
return x + y
Upvotes: 52
Reputation: 111
You can specify success and error callbacks via the link and link_err kwargs when you call apply_async. The celery docs include a clear example: http://docs.celeryproject.org/en/latest/userguide/calling.html#linking-callbacks-errbacks
Upvotes: 11