在Flask中使用Celery时出现ImportError: cannot import name App错误求助

你遇到的这个问题根源很明确：你使用的flask_celery库是针对旧版本Celery（Celery 3.x及更早）开发的，而当前安装的Celery是4.x+的新版本。新版本Celery重构了内部API，celery.app模块里已经移除了App这个类，所以才会触发导入错误。

下面给你两种可行的解决方案，推荐第一种长期维护性更好的方式：

方案一：直接整合Flask和Celery（无需依赖过时的flask_celery） #

Celery官方现在推荐直接通过自定义任务类的方式来和Flask整合，这样能避免依赖第三方过时库。修改你的代码如下：

from flask import Flask
from celery import Celery

app = Flask(__name__)
# 配置Celery的消息代理
app.config["CELERY_BROKER_URL"] = "amqp://guest:guest@localhost:5672//"
# 可选：如果需要获取任务执行结果，配置结果后端
app.config["CELERY_RESULT_BACKEND"] = "rpc://"

def make_celery(app):
    # 初始化Celery实例
    celery = Celery(
        app.import_name,
        broker=app.config['CELERY_BROKER_URL'],
        backend=app.config.get('CELERY_RESULT_BACKEND')
    )
    # 用Flask的配置更新Celery的配置
    celery.conf.update(app.config)
    
    # 自定义任务类，确保任务执行时能访问Flask应用上下文
    class ContextTask(celery.Task):
        def __call__(self, *args, **kwargs):
            with app.app_context():
                return self.run(*args, **kwargs)
    
    celery.Task = ContextTask
    return celery

celery = make_celery(app)

@celery.task(name='server.reverse')
def reverse(string):
    return string[::-1]

@app.route("/process/<name>")
def process(name):
    reverse.delay(name)
    return "I sent an async request!"

if __name__ == "__main__":
    app.run(debug=True)

关键说明： #

• 自定义的ContextTask类让Celery任务在执行时自动进入Flask的应用上下文，这样如果你的任务里需要访问Flask的配置、数据库连接等资源，都能正常工作。
• 启动Celery worker的命令要对应正确：在终端运行 celery -A server.celery worker --loglevel=info（假设你的文件名为server.py）。

方案二：安装兼容的旧版本Celery（不推荐） #

如果你一定要继续使用flask_celery库，只能安装和它兼容的Celery旧版本（比如3.1.x系列），执行以下命令：

pip install celery==3.1.26.post2

但这个方案不推荐，因为旧版本Celery存在安全漏洞，而且缺少新版本的很多实用功能，长期维护风险高。

额外注意： #

确保你已经启动了RabbitMQ服务（因为你配置的消息代理是AMQP），如果还没启动，需要先启动RabbitMQ才能让Celery正常工作。

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