python - 在celery.py中导入Django App函数

Question

我正在开发一个应用程序，我必须定期从位于我的项目目录中的 celery.py 文件调用 myapp/task.py 中的函数。但是当我从 celery.py 执行导入语句时像

这样的文件

from myapp.tasks import test

我收到以下我无法解决的错误:

django.core.exceptions.AppRegistryNotReady: Apps aren't loaded yet.

这是来自 init.py 的代码:

from __future__ import absolute_import, unicode_literals
from .celery import app as celery_app

 __all__ = ['celery_app']

celery.py 中的代码

from __future__ import absolute_import,unicode_literals
import os
from celery import Celery
from celery.schedules import crontab
from myapp.tasks import test

os.environ.setdefault('DJANGO_SETTINGS_MODULE','project.settings') app=Celery('项目')

app.config_from_object('django.conf:settings',namespace='CELERY')

app.autodiscover_tasks()

@app.on_after_configure.connect

def setup_monitor_call(sender,**kwargs):

   sender.add_periodic_task(30.0,test.s(),name='call test every 30 seconds')

现在在我的应用程序目录中有一个名为 tasks.py 的文件，其中包含我从 celery.py 调用的测试方法:

from __future__ import absolute_import, unicode_literals
from celery.decorators import task
from celery import shared_task
from celery.utils.log import get_task_logger

logger=get_task_logger(__name__)

def test():
  print("Hello Celery")

现在确切的问题是，当我在 celery.py 文件中写入以下导入语句时:

from myapp.tasks import test

然后当我运行 python manage.py runserver 时出现错误:

django.core.exceptions.AppRegistryNotReady: Apps aren't loaded yet.

我删除导入语句的那一刻，服务器运行正常。我想不通为什么？

Answer 1

此错误是由于与该导入不兼容造成的，例如循环导入。我最近在做一个 celery 项目，有时进口的东西就像纸牌屋一样脆弱。
我建议您在此示例中按如下方式调用任务:

.....otherimports
from celery import Celery

import incidencias.celery_conf as celery_conf

app = Celery('myname')
app.config_from_object(celery_conf, namespace='CELERY')
app.conf.timezone = 'Europe/Madrid'

def some_def():
     code code code code....  
     app.send_task('myapp.tasks.test', args=[whatyouneed.ifneed])