I am trying to retrieve all currently pending and active tasks from Celery.
active, scheduled and reserved are simple to obtain:
i = celery_app.control.inspect()
tasks=list()
for k in i.active():
tasks += i.active()[k]
for k in i.reserved():
tasks += i.reserved()[k]
for k in i.scheduled():
tasks += i.scheduled()[k]
However, this does not include tasks that have not been scheduled or reserved yet. The following code retrieves these tasks, but obviously only works with Redis, assumes a JSON serialization, and it's not ideal for a variety of reasons.
r = redis.StrictRedis(
host='localhost',
port='6379',
)
tasks = [json.loads(pending_task)['headers'] for pending_task in r.lrange('celery', 0, -1)]
Is there a way to retrieve these pending tasks in a form similar to inspect.active() with easy access to args, kwargs, id (preferably through Celery, so I do not need to specify the Celery backend and serialization, etc)?
EDIT (hedging my bet)
From what I can tell this is not simple, and there is no i.pending() in celery. So, I have a few sub questions which will allow me to write this myself.
- How to get a list of active celery queues? (the above assumes default queue only) Perhaps something like
queues = [q['exchange']['name'] for host in i.active_queues().values() for q in host]) - How to generically get messages in those queues? This would be a generic version of
r.lrange('celery', 0, -1)above. How to generically decode/de-serialize messages? This would be a generic version ofjson.loads(...) abovecelery_app.backend.decode
With that info it should be straightforward to pull the necessary information from the message headers.