无法将模型导入 Django 中的 Celery 任务

Question

我正在尝试配置 Django 和 Celery，当我将任务导入 models.py 文件并同时将模型导入 tasks.py 文件时遇到问题。芹菜在其他方面工作。请参阅下面的代码...

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'

这是错误信息：

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

你有一个循环依赖问题。

最好的做法是重构您的应用程序，使users 不依赖于core，而core 也依赖于users。如果不了解您的特定领域，很难就如何做到这一点给出明确的答案。

你也可以通过在函数执行时导入交叉依赖来避免错误：

# 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)

这将在短期内解决您的问题，但我不建议长期这样做，因为交叉依赖项表明您在建模域数据和逻辑方面可能存在缺陷。

Answer 2

确实看起来像一个循环依赖，但有时它是不可避免的，所以你无法避免它。

通常在这些情况下，将其导入函数可能会有所帮助（请参阅 azundo 的评论），但通常这不起作用。

Django 有一个内置的应用程序注册表，您可以访问它来检索这种情况下加载的模型，请参阅：

# 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()

有关更多信息，请参阅Django refs，您还可以添加更多参数以使其专门针对您的情况工作。