[Django]-Detect whether Celery is Available/Running

From the documentation of celery 4.2:

from your_celery_app import app


def get_celery_worker_status():
    i = app.control.inspect()
    availability = i.ping()
    stats = i.stats()
    registered_tasks = i.registered()
    active_tasks = i.active()
    scheduled_tasks = i.scheduled()
    result = {
        'availability': availability,
        'stats': stats,
        'registered_tasks': registered_tasks,
        'active_tasks': active_tasks,
        'scheduled_tasks': scheduled_tasks
    }
    return result

of course you could/should improve the code with error handling…

To check the same using command line in case celery is running as daemon,

• Activate virtualenv and go to the dir where the ‘app’ is
• Now run : celery -A [app_name] status
• It will show if celery is up or not plus no. of nodes online

Source:
http://michal.karzynski.pl/blog/2014/05/18/setting-up-an-asynchronous-task-queue-for-django-using-celery-redis/

The following worked for me:

import socket
from kombu import Connection

celery_broker_url = "amqp://localhost"

try:
    conn = Connection(celery_broker_url)
    conn.ensure_connection(max_retries=3)
except socket.error:
    raise RuntimeError("Failed to connect to RabbitMQ instance at {}".format(celery_broker_url))

One method to test if any worker is responding is to send out a ‘ping’ broadcast and return with a successful result on the first response.

from .celery import app  # the celery 'app' created in your project

def is_celery_working():
    result = app.control.broadcast('ping', reply=True, limit=1)
    return bool(result)  # True if at least one result

This broadcasts a ‘ping’ and will wait up to one second for responses. As soon as the first response comes in, it will return a result. If you want a False result faster, you can add a timeout argument to reduce how long it waits before giving up.

I found an elegant solution:

from .celery import app
try:
    app.broker_connection().ensure_connection(max_retries=3)
except Exception as ex:
    raise RuntimeError("Failed to connect to celery broker, {}".format(str(ex)))

You can use ping method to check whether any worker (or specific worker) is alive or not
https://docs.celeryproject.org/en/latest/_modules/celery/app/control.html#Control.ping

celey_app.control.ping()

The below script is worked for me.

    #Import the celery app from project
    from application_package import app as celery_app
    def get_celery_worker_status():
        insp = celery_app.control.inspect()
        nodes = insp.stats()
        if not nodes:
            raise Exception("celery is not running.")
        logger.error("celery workers are: {}".format(nodes))
        return nodes

You can test on your terminal by running the following command.

celery -A proj_name worker -l INFO

You can review every time your celery runs.

Run celery status to get the status.

When celery is running,

(venv) ubuntu@server1:~/project-dir$ celery status
->  celery@server1: OK

1 node online.

When no celery worker is running, you get the below information displayed in terminal.

(venv) ubuntu@server1:~/project-dir$ celery status
Error: No nodes replied within time constraint