How to Check Task Status in Celery

How to check task status in Celery?

Every Task object has a .request property, which contains it AsyncRequest object. Accordingly, the following line gives the state of a Task task:

task.AsyncResult(task.request.id).state

Python Celery get task status

AsyncResult is an object and state property is what you want:

@app.route("/checkStatus", methods=['GET'])
def checkStatus():
  uuid = request.args.get('uuid') 
  res = removeUser.AsyncResult(uuid)
  print(uuid)
  if res.state == "SUCCESS":
    return "success"
  else:
    return "progress"

Notes:

1. Task might fail as well so that there's a better way to check if the task has executed (rather than compare to "SUCCESS").
2. If you'll add the celery app you don't have to use a specific task and this function can help you to get the state of any celery task by uuid.

@app.route("/checkStatus", methods=['GET'])
def checkStatus():
  uuid = request.args.get('uuid') 
  res = AsyncResult(uuid, app=client)
  print(uuid)
  if res.ready():
    return "success"
  else:
    return "progress"

How to Get the Task Status and Result of a Celery Task Type When Multiple Tasks Are Defined?

id and state are just properties of the AsyncResult objects. If you looked at documentation for the AsyncResult class, you would find the name property which is exactly what you are asking for.

Celery task status is only 'PENDING' and 'SUCCESS'.Why is there no STARTED in the task status?

That feature is not enabled by default. More about it here: https://docs.celeryproject.org/en/latest/userguide/tasks.html#started

Check status of Celery worker

you can see active workers using flower API or by directly querying celery using below manner:

from celery import Celery
app = Celery('vwadaptor',
             broker='redis://workerdb:6379/0',
             backend='redis://workerdb:6379/0')

app.control.inspect().active()

Shows you the active workers and their currently executing jobs.

Retrieve task result by id in Celery

It works using AsyncResult. (see this answer)

So first create the task:

from cel.tasks import add

res = add.delay(3,4)
print(res.status) # 'SUCCESS'
print(res.id) # '432890aa-4f02-437d-aaca-1999b70efe8d'

Then start another python shell:

from celery.result import AsyncResult
from cel.tasks import app

res = AsyncResult('432890aa-4f02-437d-aaca-1999b70efe8d',app=app)

print(res.state) # 'SUCCESS'
print(res.get()) # 7

---

Related Topics