Celery - 在tasks.py中导入模型

Question

我在访问tasks.py中的模型时遇到问题

\n\n

我的目标是在应用程序的各个部分发送电子邮件（用户注册、重置密码等）。为此，我将用户 ID 传递给名为“send_email”的 celery 任务。

\n\n

@shared_task()\ndef send_email(sender_id=None, receiver_id=None, type=None, message=None):\n\n    sender = User.objects.get(id=sender_id)\n    receiver = User.objects.get(id=receiver_id)\n\n    logger.info("Starting send email")\n    Email.send_email(sender, receiver, type, message)\n    logger.info("Finished send email")\n

\n\n

然后，该任务需要使用该 ID 来检索用户并向他们发送电子邮件。当尝试将 User 模型导入到tasks.py 文件中时，这会失败。

\n\n

我收到一个错误

\n\n

raise AppRegistryNotReady("Apps aren\'t loaded yet.") django.core.exceptions.AppRegistryNotReady: Apps aren\'t loaded yet.\n

\n\n

我尝试过的事情

\n\n

\n

1. 在tasks.py文件顶部调用 django.setup() - 原因

   \n\n

   raise RuntimeError("populate() isn\'t reentrant") \n

   \n\n

   当放入 send_email 方法时也会导致相同的错误。这些是对 SO 中其他类似问题的建议

\n

2. 在“send_email”方法中导入模型，允许工作程序启动，但会导致以下错误

   \n\n

   raise AppRegistryNotReady("Apps aren\'t loaded yet.") \n

   \n\n

   这是SO中类似问题的另一个建议

\n

3. 在调用有效的“send_email”函数时删除.delay（在tasks.py文件顶部或send_email方法中导入），但由于任务不再异步，因此没有任何好处，但可能会缩小问题范围？

\n

\n\n

注意事项？

\n\n

\n
1. 我使用扩展 AbstractBaseUser 的自定义用户模型，我在 celery 中看到了许多与此相关的 github 问题，但我相信这些问题应该在 celery v3.1 中修复
\n

2. 我使用 celery v4.1、django 1.11.10、python 2.7，并使用 RabbitMQ 作为代理并在虚拟环境上运行工作程序/服务器。我正在开始使用我的工作人员

   \n\n

   celery -A api worker -l info\n

   \n\n

   在终端窗口上，然后使用 pycharm 的终端来启动服务器

   \n\n

   python manage.py runserver\n

   \n\n

   那么实际上有 2 个环境吗？这可能是问题所在吗？

\n

3. 这可能相关，也可能不相关，但为了让我的自定义用户模型在我的 app/models.py 中工作，我只有一行导入用户模型，否则我会得到一个

   \n\n

   django.core.exceptions.ImproperlyConfigured: AUTH_USER_MODEL refers to model \'myApi.User\' that has not been installed\n

\n

4. 我尝试将身份验证模型设置为 \'myApi.user.User\' （用户是声明模型的文件夹，但得到一个

   \n\n

   Invalid model reference \'myApi.user.User\'. String model references must be of the form \'app_label.ModelName\'.\n

   \n\n

   所以我猜这就是为什么需要在 myApi/models.py 中导入以便可以在此处获取它的原因？

\n

\n\n

项目结构

\n\n

\xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 api\n\xe2\x94\x82\xc2\xa0\xc2\xa0 \xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 __init__.py\n\xe2\x94\x82\xc2\xa0\xc2\xa0 \xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 settings.py\n\xe2\x94\x82\xc2\xa0\xc2\xa0 \xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 urls.py\n\xe2\x94\x82\xc2\xa0\xc2\xa0 \xe2\x94\x94\xe2\x94\x80\xe2\x94\x80 wsgi.py\n\xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 celerySettings.py # my celery.py\n\xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 db.sqlite3\n\xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 myApi\n\xe2\x94\x82\xc2\xa0\xc2\xa0 \xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 __init__.py\n\xe2\x94\x82\xc2\xa0\xc2\xa0 \xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 admin.py\n\xe2\x94\x82\xc2\xa0\xc2\xa0 \xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 apps.py\n\xe2\x94\x82\xc2\xa0\xc2\xa0 \xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 tasks.py\n\xe2\x94\x82\xc2\xa0\xc2\xa0 \xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 urls.py\n\xe2\x94\x82\xc2\xa0\xc2\xa0 \xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 user\n\xe2\x94\x82\xc2\xa0\xc2\xa0 \xe2\x94\x82\xc2\xa0\xc2\xa0 \xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 __init__.py\n\xe2\x94\x82\xc2\xa0\xc2\xa0 \xe2\x94\x82\xc2\xa0\xc2\xa0 \xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 managers.py\n\xe2\x94\x82\xc2\xa0\xc2\xa0 \xe2\x94\x82\xc2\xa0\xc2\xa0 \xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 models.py\n\xe2\x94\x82\xc2\xa0\xc2\xa0 \xe2\x94\x82\xc2\xa0\xc2\xa0 \xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 serializers.py\n\xe2\x94\x82\xc2\xa0\xc2\xa0 \xe2\x94\x82\xc2\xa0\xc2\xa0 \xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 urls.py\n\xe2\x94\x82\xc2\xa0\xc2\xa0 \xe2\x94\x82\xc2\xa0\xc2\xa0 \xe2\x94\x94\xe2\x94\x80\xe2\x94\x80 views.py\n\xe2\x94\x82\xc2\xa0\xc2\xa0 \xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 utils\n\xe2\x94\x82\xc2\xa0\xc2\xa0 \xe2\x94\x82\xc2\xa0\xc2\xa0 \xe2\x94\x94\xe2\x94\x80\xe2\x94\x80 Email.py\n\xe2\x94\x82\xc2\xa0\xc2\xa0 \xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 views.py\n\xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 manage.py\n\xe2\x94\x94\xe2\x94\x80\xe2\x94\x80 static\n

\n\n

任务.py

\n\n

from __future__ import absolute_import, unicode_literals\nfrom celery.schedules import crontab\nfrom celery.task import periodic_task\nfrom celery.utils.log import get_task_logger\nfrom celery import shared_task\n\nfrom celery import current_app\n\nfrom .user.models import User\nfrom .utils import Email\n\nlogger = get_task_logger(__name__)\n\n@shared_task()\ndef send_email(sender_id=None, receiver_id=None, type=None, message=None):\n\n    sender = User.objects.get(id=sender_id)\n    receiver = User.objects.get(id=receiver_id)\n\n    logger.info("Starting send email")\n    Email.send_email(sender, receiver, type, message)\n    logger.info("Finished send email")\n

\n\n

设置.py

\n\n

....\nINSTALLED_APPS = [\n    \'rest_framework\',\n    \'django.contrib.admin\',\n    \'django.contrib.auth\',\n    \'django.contrib.contenttypes\',\n    \'django.contrib.sessions\',\n    \'django.contrib.messages\',\n    \'django.contrib.staticfiles\',\n    \'corsheaders\',\n    \'myApi\',\n    \'celery\',\n    \'rest_framework.authtoken\',\n    \'rest_framework.renderers\',\n]\n\nAUTH_USER_MODEL = \'myApi.User\'\nCELERY_IMPORTS = (\'api.myApi.tasks\')\n....\n

\n\n

celerySettings.py

\n\n

from __future__ import absolute_import, unicode_literals\nfrom django.conf import settings\nimport os\nfrom celery import Celery\n\n\n# set the default Django settings module for the \'celery\' program.\nos.environ.setdefault(\'DJANGO_SETTINGS_MODULE\', \'api.api.settings\')\n\napp = Celery(\'api\', broker=\'amqp://\')\n\n# Using a string here means the worker doesn\'t have to serialize\n# the configuration object to child processes.\n# - namespace=\'CELERY\' means all celery-related configuration keys\n#   should have a `CELERY_` prefix.\napp.config_from_object(settings, namespace=\'CELERY\')\n\n# Load task modules from all registered Django app configs.\napp.autodiscover_tasks()\n\n@app.task(bind=True)\ndef debug_task(self):\n    print(\'Request: {0!r}\'.format(self.request))\n

\n\n

myApi/models.py

\n\n

from user.models import User\n

\n\n

myApi/admin.py

\n\n

# -*- coding: utf-8 -*-\nfrom __future__ import unicode_literals\n\nfrom django.contrib import admin\n\nfrom user.models import User\n\nadmin.site.register(User)\n

\n\n

api/wsgi.py

\n\n

import os\n\nfrom django.core.wsgi import get_wsgi_application\n\nos.environ.setdefault("DJANGO_SETTINGS_MODULE", "api.settings")\n\napplication = get_wsgi_application()\n

\n\n

任何建议将不胜感激。也对长篇文章感到抱歉，这是我的第一篇文章，所以不确定需要多少细节。

\n

Answer 1

我发现了我的问题。如果它可以帮助其他人解决这个问题，我需要添加这一行

sys.path.append(os.path.abspath('api'))

在我的 celerySettings.py 中用于选择模型。

所以现在看起来像这样

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

sys.path.append(os.path.abspath('api'))

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

app = Celery('api', broker='amqp://')

# Using a string here means the worker doesn't have to serialize
# the configuration object to child processes.
# - namespace='CELERY' means all celery-related configuration keys
#   should have a `CELERY_` prefix.
app.config_from_object(settings, namespace='CELERY')

# Load task modules from all registered Django app configs.
app.autodiscover_tasks()

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

然后，当我实际尝试在本地查询数据库中的模型时，遇到了另一个问题。Celery 说我的数据库表不存在，这是因为它正在创建一个新数据库，位于实际本地数据库文件所在文件夹的上方，要修复它，我只需更改数据库名称

"db.sqlite3"

到

os.path.join(os.path.dirname(__file__), "db.sqlite3")

在设置.py

这有效地改变了它

api/db.sqlite3

芹菜用

希望这对其他人有帮助，因为我花了太多时间来解决这个问题。