My custom django-admin command won't call celery task

Question

I am trying to write a custom django-admin command that executes a celery task, however the task doesn't execute and django just hangs when I try.

from django.core.management.base import BaseCommand
from myapp.tasks import my_celery_task

class Command(BaseCommand):
    def handle(self, *args, **options):
        print "starting task"
        my_celery_task.delay()
        print "task has been sent"

The output I receive when calling the command is:

starting task

I never reach the "task has been sent" line. It just hangs. I'm not sure why the task isn't running. Celery tasks are called perfectly when called by a view.

Answer 1

This seems to work for me with celery 5.1.2, but not if it is a shared_task

====== tasks.py

from celery import current_app
from celery.schedules import crontab
app = current_app._get_current_object()

@app.task
def my_celery_task():
    # Do something interesting
    return 0 # Return the result of this operation

====== my_command.py

from accounts.tasks import my_celery_task
...
    def handle(self, *args, **options):
        self.stdout.write("Queing request")
        result = my_celery_task.delay()
        self.stdout.write("Waiting for task to end")
        updated = result.wait(timeout=120.0)

        self.stdout.write(f"Returned {updated}")

And, if I put the task in the celery.py file I ended up with a deadlock which was a PITA to debug