Avoiding duplicate tasks in celery broker

Question

I want to create the following flow using celery configuration\api:

• Send TaskA(argB) Only if celery queue has no TaskA(argB) already pending

Is it possible? how?

Answer 1

You can make your job aware of other tasks by some sort of memoization. If you use a cache control key (redis, memcached, /tmp, whatever is handy), you can make execution depend on that key. I'm using redis as an example.

from redis import Redis

@app.task
def run_only_one_instance(params):
    try:
        sentinel =  Redis().incr("run_only_one_instance_sentinel")
        if sentinel == 1:
            #I am the legitimate running task
            perform_task()
        else:
            #Do you want to do something else on task duplicate?
            pass
        Redis().decr("run_only_one_instance_sentinel")
    except Exception as e:
        Redis().decr("run_only_one_instance_sentinel")
        # potentially log error with Sentry?
        # decrement the counter to insure tasks can run
        # or: raise e

Answer 2

I cannot think of a way but to

1. Retrieve all executing and scheduled tasks via celery inspect

2. Iterate through them to see if your task is there.

check this SO question to see how the first point is done.

good luck