Flask集成Celery遇循环导入错误：无法从workers导入celery

错误根源 #

你遇到的ImportError是导入顺序混乱+wsgi.py错误代码导致的：

1. wsgi.py中错误地将已实例化的Flask app当作函数调用：app = app(celery=celery)，这行代码会破坏原有app对象，还会在导入过程中触发意外的依赖加载。
2. 启动Celery Worker时，workers模块加载make_celery，若wsgi.py被间接导入，会形成循环依赖：wsgi导入workers的celery，workers导入celery_app，celery_app导入core的app，wsgi同时也导入core的app，最终导致workers模块还未完成celery初始化就被wsgi导入，触发错误。

修复步骤 #

1. 修正wsgi.py的错误代码 #

从core导入的app已经是Flask实例，不能像工厂函数一样调用。修改后的wsgi.py：

from core import app
# 若需在Flask中使用celery，直接导入即可，无需重新赋值app
from workers import celery

if __name__ == '__main__':
    app.run()

2. 优化Celery初始化逻辑（更健壮） #

为彻底避免导入问题，将Celery初始化与Flask应用解耦，用Flask配置统一管理Celery参数：

修改celery_app.py

from celery import Celery

def make_celery(app=None):
    # 延迟导入core的app，避免提前触发依赖加载
    if not app:
        from core import app
    
    celery = Celery(
        app.import_name,
        backend=app.config.get('CELERY_BACKEND', "amqp://guest:guest@localhost:15672/BACKEND"),
        broker=app.config.get('CELERY_BROKER_URL', "amqp://guest:guest@localhost:15672/BROKER")
    )
    celery.conf.update(app.config)

    class ContextTask(celery.Task):
        def __call__(self, *args, **kwargs):
            with app.app_context():
                return self.run(*args, **kwargs)

    celery.Task = ContextTask
    return celery

（可选）在core/init.py中添加Celery配置

把Celery参数统一放到Flask配置里，方便管理：

from flask import Flask
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
# 配置Celery连接信息
app.config['CELERY_BROKER_URL'] = "amqp://guest:guest@localhost:15672/BROKER"
app.config['CELERY_BACKEND'] = "amqp://guest:guest@localhost:15672/BACKEND"

db = SQLAlchemy(app)

3. 正确启动Celery Worker #

在flask目录下激活虚拟环境后，执行以下命令：

source venv/bin/activate
celery -A workers worker --loglevel=info

4. 启动Flask应用 #

运行wsgi.py验证：

python wsgi.py

额外注意事项 #

• 确保RabbitMQ服务已启动（你使用AMQP作为broker和backend），否则Celery无法正常连接。
• 不要在core/init.py中导入workers或celery相关模块，防止触发循环依赖。
• 若需在Flask视图中调用Celery任务，直接从workers导入celery实例即可。

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