python - Find out whether celery task exists

Question

Is it possible to find out whether a task with a certain task id exists? When I try to get the status, I will always get pending.

>>> AsyncResult('...').status
'PENDING'

I want to know whether a given task id is a real celery task id and not a random string. I want different results depending on whether there is a valid task for a certain id.

There may have been a valid task in the past with the same id but the results may have been deleted from the backend.

Answer 1

Celery does not write a state when the task is sent, this is partly an optimization (see the documentation).

If you really need it, it's simple to add:

from celery import current_app
# `after_task_publish` is available in celery 3.1+
# for older versions use the deprecated `task_sent` signal
from celery.signals import after_task_publish

# when using celery versions older than 4.0, use body instead of headers

@after_task_publish.connect
def update_sent_state(sender=None, headers=None, **kwargs):
    # the task may not exist if sent using `send_task` which
    # sends tasks by name, so fall back to the default result backend
    # if that is the case.
    task = current_app.tasks.get(sender)
    backend = task.backend if task else current_app.backend
 
    backend.store_result(headers['id'], None, "SENT")

Then you can test for the PENDING state to detect that a task has not (seemingly) been sent:

>>> result.state != "PENDING"