Check Celery Queues

queues.py

from django.conf import settings

used_queues = set()
 
for route, q_dict in settings.CELERY_ROUTES.items():
   module, task = route.rsplit('.', 1)
   mod = __import__(module, globals(), locals(), [task])
   try:
     t = getattr(mod, task)
   except AttributeError:
     print "TASK ROUTED WITH NO ACTUAL METHOD", task
   else:
    if not hasattr(t, 'AsyncResult'):
        print "METHOD ROUTED BUT IS NOT A TASK", task
    else:
        used_queues.add(q_dict['queue'])

defined_queues = set(settings.CELERY_QUEUES.keys())

unused_queues = defined_queues - used_queues
undefined_queues = used_queues - defined_queues
print "QUEUES WITH NO TASKS"
for q in unused_queues:
    print q

print "QUEUES NOT EXPLICITY DEFINED"
for q in undefined_queues:
    print q