如何解决Celery处理Django自定义模型时的AppRegistryNotReady错误

问题根源 #

在Celery实例初始化完成前，你直接在celery.py中导入了包含Django模型的tasks模块，此时Django的应用注册表还未加载完成，触发模型导入时就会抛出AppRegistryNotReady错误。同时，你在beat_schedule中直接调用了任务函数savings_increment()，这也是错误的写法。

修复步骤 #

1. 修改celery.py，移除提前导入的tasks并修正beat_schedule配置

把celery.py中对tasks的直接导入删除，同时将beat_schedule里的任务指定方式改为任务的字符串路径（而非直接调用函数）：

from __future__ import absolute_import, unicode_literals
import os

from celery import Celery
from celery.schedules import crontab
from django.conf import settings

from .load_env import load_env

load_env()

# 设置Celery默认的Django配置模块
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "agric360.settings")

app = Celery("agric360")
app.config_from_object("django.conf:settings", namespace="CELERY")
app.conf.beat_schedule={
    'increment':{
        # 使用任务的完整字符串路径，不要调用函数
        'task': 'ecoperative.tasks.savings_increment',
        'schedule': crontab(minute=1)
    }
}

# 自动发现所有app中的tasks.py
app.autodiscover_tasks()

2. 保持tasks.py不变

你的tasks.py代码本身没有问题，autodiscover_tasks()会在Celery初始化完成、Django应用加载好之后，自动找到并加载这个文件，此时模型导入就不会触发错误。

原理说明 #

• Celery的beat_schedule需要的是任务的唯一标识字符串（格式为app_name.tasks.task_function_name），而非直接执行任务函数。
• app.autodiscover_tasks()会自动扫描所有已注册Django应用中的tasks.py文件，在Django应用注册表加载完成后才会导入这些文件，避免了提前导入模型导致的错误。

内容的提问来源于stack exchange，提问作者miracle godwin