check_celery.py

from datetime import timedelta
from time import sleep
from celery.task import task

from django.core.management.base import BaseCommand, CommandError
from django.utils import timezone


@task
def add(x, y):
    return x + y


class Command(BaseCommand):
    help = "Check if celery is up and running."

    def handle(self, *args, **options):
        try:
            result = add.apply_async(
                args=[4, 4], expires=timezone.now() + timedelta(seconds=3), connect_timeout=3)
            now = timezone.now()
            while (now + timedelta(seconds=3)) > timezone.now():
                if result.result == 8:
                    return "success"
                sleep(0.5)
            return "failure"
        except Exception as e:
            raise CommandError("check_celery_status: Celery Service Error", e)