Retrieve list of tasks in a queue in Celery

Question

How can I retrieve a list of tasks in a queue that are yet to be processed

User · Answer

A copy-paste solution for Redis with json serialization:

def get_celery_queue_items(queue_name):
    import base64
    import json  

    # Get a configured instance of a celery app:
    from yourproject.celery import app as celery_app

    with celery_app.pool.acquire(block=True) as conn:
        tasks = conn.default_channel.client.lrange(queue_name, 0, -1)
        decoded_tasks = []

    for task in tasks:
        j = json.loads(task)
        body = json.loads(base64.b64decode(j['body']))
        decoded_tasks.append(body)

    return decoded_tasks

It works with Django. Just don't forget to change yourproject.celery.

User · Answer

from celery task control import inspect def key in list k  l       return bool  True for i in l if k in i values      def check task task id       task value dict   inspect   active   values       for task list in task value dict          if self key in list task id  task list                return True     return False

User · Answer

This worked for me in my application  def get celery queue active jobs queue name       connection    lt CELERY APP INSTANCE gt  connection        try          channel   connection channel           name  jobs  consumers   channel queue declare queue queue name  passive True          active jobs               def dump message message               active jobs append message properties  application headers    task             channel basic consume queue queue name  callback dump message           for job in range jobs               connection drain events            return active jobs     finally          connection close    active jobs will be a list of strings that correspond to tasks in the queue  Don t forget to swap out CELERY APP INSTANCE with your own  Thanks to  ashish for pointing me in the right direction with his answer here  https   stackoverflow com a 19465670 9843399

User · Answer

If you don t use prioritized tasks  this is actually pretty simple if you re using Redis  To get the task counts   redis-cli -h HOST -p PORT -n DATABASE NUMBER llen QUEUE NAME   But  prioritized tasks use a different key in redis  so the full picture is slightly more complicated  The full picture is that you need to query redis for every priority of task  In python  and from the Flower project   this looks like   PRIORITY SEP     x06 x16  DEFAULT PRIORITY STEPS    0  3  6  9    def make queue name for pri queue  pri          Make a queue name for redis      Celery uses PRIORITY SEP to separate different priorities of tasks into     different queues in Redis  Each queue-priority combination becomes a key in     redis with names like        - batch1 x06 x163  lt -- P3 queue named batch1      There s more information about this in Github  but it doesn t look like it      will change any time soon         - https   github com celery kombu issues 422      In that ticket the code below  from the Flower project  is referenced         - https   github com mher flower blob master flower utils broker py L135       param queue  The name of the queue to make a name for       param pri  The priority to make a name with       return  A name for the queue-priority pair              if pri not in DEFAULT PRIORITY STEPS          raise ValueError  Priority not in priority steps       return   0  1  2   format    queue  PRIORITY SEP  pri  if pri else                                  queue              def get queue length queue name  celery           Get the number of tasks in a celery queue        param queue name  The name of the queue you want to inspect       return  the number of items in the queue              priority names    make queue name for pri queue name  pri  for pri in                       DEFAULT PRIORITY STEPS      r   redis StrictRedis          host settings REDIS HOST          port settings REDIS PORT          db settings REDIS DATABASES  CELERY              return sum  r llen x  for x in priority names     If you want to get an actual task  you can use something like   redis-cli -h HOST -p PORT -n DATABASE NUMBER lrange QUEUE NAME 0 -1   From there you ll have to deserialize the returned list  In my case I was able to accomplish this with something like   r   redis StrictRedis      host settings REDIS HOST      port settings REDIS PORT      db settings REDIS DATABASES  CELERY      l   r lrange  celery   0  -1  pickle loads base64 decodestring json loads l 0    body       Just be warned that deserialization can take a moment  and you ll need to adjust the commands above to work with various priorities

User · Answer

As far as I know Celery does not give API for examining tasks that are waiting in the queue  This is broker-specific  If you use Redis as a broker for an example  then examining tasks that are waiting in the celery  default  queue is as simple as    connect to the broker database list items in the celery list  LRANGE command for an example    Keep in mind that these are tasks WAITING to be picked by available workers  Your cluster may have some tasks running - those will not be in this list as they have already been picked

User · Answer

I think the only way to get the tasks that are waiting is to keep a list of tasks you started and let the task remove itself from the list when it s started   With rabbitmqctl and list queues you can get an overview of how many tasks are waiting  but not the tasks itself  http   www rabbitmq com man rabbitmqctl 1 man html  If what you want includes the task being processed  but are not finished yet  you can keep a list of you tasks and check their states   from tasks import add result   add delay 4  4   result ready     True if finished   Or you let Celery store the results with CELERY RESULT BACKEND and check which of your tasks are not in there

User · Answer

I ve come to the conclusion the best way to get the number of jobs on a queue is to use rabbitmqctl as has been suggested several times here  To allow any chosen user to run the command with sudo I followed the instructions here  I did skip editing the profile part as I don t mind typing in sudo before the command    I also grabbed jamesc s grep and cut snippet and wrapped it up in subprocess calls    from subprocess import Popen  PIPE p1   Popen   sudo    rabbitmqctl    list queues    -p     name of your virtula host    stdout PIPE  p2   Popen   grep    -e     celery s    stdin p1 stdout  stdout PIPE  p3   Popen   cut    -f2    stdin p2 stdout  stdout PIPE  p1 stdout close   p2 stdout close   print  number of jobs on queue   i    int p3 communicate   0

User · Answer

if you are using rabbitMQ  use this in terminal   sudo rabbitmqctl list queues   it will print list of queues with number of pending tasks  for example   Listing queues     0b27d8c59fba4974893ec22d478a7093    0 0e0a2da9828a48bc86fe993b210d984f    0 10 torob2 celery pidbox 0 11926b79e30a4f0a9d95df61b6f402f7    0 15c036ad25884b82839495fb29bd6395    1 celerey mail worker torob2 celery pidbox    0 celery  166 celeryev 795ec5bb-a919-46a8-80c6-5d91d2fcf2aa   0 celeryev faa4da32-a225-4f6c-be3b-d8814856d1b6   0   the number in right column is number of tasks in the queue  in above  celery queue has 166 pending task

User · Answer

EDIT  See other answers for getting a list of tasks in the queue   You should look here  Celery Guide - Inspecting Workers  Basically this   from celery app control import Inspect    Inspect all nodes  i   Inspect      Show the items that have an ETA or are scheduled for later processing i scheduled      Show tasks that are currently active  i active      Show tasks that have been claimed by workers i reserved     Depending on what you want

User · Answer

The celery inspect module appears to only be aware of the tasks from the workers perspective  If you want to view the messages that are in the queue  yet to be pulled by the workers  I suggest to use pyrabbit  which can interface with the rabbitmq http api to retrieve all kinds of information from the queue   An example can be found here  Retrieve queue length with Celery  RabbitMQ  Django

User · Answer

If you control the code of the tasks then you can work around the problem by letting a task trigger a trivial retry the first time it executes  then checking inspect   reserved    The retry registers the task with the result backend  and celery can see that  The task must accept self or context as first parameter so we can access the retry count    task bind True  def mytask self       if self request retries    0          raise self retry exc MyTrivialError    countdown 1            This solution is broker agnostic  ie  you don t have to worry about whether you are using RabbitMQ or Redis to store the tasks   EDIT  after testing I ve found this to be only a partial solution  The size of reserved is limited to the prefetch setting for the worker

User · Answer

If you are using Celery Django simplest way to inspect tasks using commands directly from your terminal in your virtual environment or using a full path to celery   Doc  http   docs celeryproject org en latest userguide workers html highlight revoke inspecting-workers    celery inspect reserved   celery inspect active   celery inspect registered   celery inspect scheduled   Also if you are using Celery RabbitMQ you can inspect the list of queues using the following command   More info  https   linux die net man 1 rabbitmqctl    sudo rabbitmqctl list queues

User · Answer

To retrieve tasks from backend  use this  from amqplib import client 0 8 as amqp conn   amqp Connection host  localhost 5672    userid  guest                          password  guest   virtual host      insist False  chan   conn channel   name  jobs  consumers   chan queue declare queue  queue name   passive True

User · Answer

With subprocess run    import subprocess import re active process txt   subprocess run   celery    -A    my proj    inspect    active                                            stdout subprocess PIPE  stdout decode  utf-8   return len re findall r worker pid   active process txt     Be careful to change my proj with your proj

[python] Retrieve list of tasks in a queue in Celery

Examples related to python

Examples related to celery