Cannot import models into Celery tasks in Django

Question

I am trying to configure Django and Celery and am running into an issue when I import the task into the models.py file and simultaneously import a model into the tasks.py file. Celery is otherwise working. See code below...

core/celery.py

from __future__ import absolute_import, unicode_literals
import os
from celery import Celery
from django.conf import settings


# set the default Django settings module for the 'celery' program.
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'core.settings')

app = Celery('core')
app.config_from_object('django.conf:settings')
app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)


@app.task(bind=True)
def debug_task(self):
    print('Request: {0!r}'.format(self.request))


users/models.py

from django.db import models
from core.tasks import celery_test #this triggers the error

class CustomUser(AbstractUser):
    username = None
    email = models.EmailField(_('email address'), unique=True)


    def __str__(self):
        return self.email


core/tasks.py

from celery.decorators import task
from users.models import CustomUser

@task(name="celery_test_task")
def celery_test_task():
    print(CustomUser)

core/settings/base.py

BROKER_URL = 'redis://localhost:6379'
CELERY_RESULT_BACKEND = 'redis://localhost:6379'
CELERY_ACCEPT_CONTENT = ['application/json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_TIMEZONE = 'Africa/Nairobi'

Here is the error message:

File "...\src\users\models.py", line 6, in <module>
    from core.tasks import celery_test
File "...\src\core\tasks.py", line 2, in <module>
    from users.models import CustomUser
ImportError: cannot import name 'CustomUser' from 'users.models' (...\src\users\models.py)

Answer 1

Indeed it seems like a circular dependency, but sometimes it is inevitable, so you cannot avoid it.

Quite often in these cases, importing it in the function may help (see azundo's comment), but usually that does not work.

Django has a build in app registry which you can access to retrieve loaded models for this case, see:

# core/tasks.py

from celery.decorators import task
from django.apps import apps

@task(name="celery_test_task") 
def celery_test_task():
    # call apps via Django
    model = apps.get_model(app_label='users', model_name='CustomUser')
    # Now models is accessible, such as model.objects.get_or_create()

See Django refs for more info, you can also add some more arguments to make it work specifically for your case.

Answer 2

You have a circular dependency problem.

The best course of action would be to restructure your apps so that users doesn't depend on core while core also depends on users. It's hard to give a clear answer on how to do this without understanding your particular domain.

You can also avoid the error by importing the cross dependency at function execution time:

# core/tasks.py

from celery.decorators import task

@task(name="celery_test_task")
def celery_test_task():
    # move import to function runtime instead of module definition time
    from users.models import CustomUser
    print(CustomUser)

This will solve your problem short-term but I would not recommend doing this long-term as cross-dependencies suggest that you likely have a flaw in how you're modeling your domain data and logic.