Celery worker进程不加载Flask应用实例,导致current_app、db等依赖应用上下文的对象不可用;正确做法是显式推送应用上下文,如通过ContextTask或create_app()配合app.app_context()。

Celery worker 进程启动时根本不会加载 Flask 应用实例,current_app 和 db 在任务函数里不是“配置错了”,而是压根不存在。
worker 进程和 Flask 应用是两个完全隔离的 Python 进程
Celery worker 启动后只认 celery 实例,不执行 app = Flask(__name__),也不调 create_app()。所以:
– current_app 是未绑定的代理对象,一访问就报 RuntimeError: Working outside of application context
– db.session 虽然能 import,但底层没上下文支撑,调 .add() 或 .commit() 必崩
– 所有依赖应用上下文的扩展(如 cache、mail)同理失效
celery -A 指向错误导致 import 失败或上下文为空
celery -A 后面的模块路径必须满足两个条件:能被 Python 成功导入,且最终暴露一个已配置好 broker/backend 的 celery 实例。
常见问题包括:
– 把 celery_app.py 放在 app/ 目录下却漏了 app/__init__.py,导致无法作为包导入
– 写 celery -A app.celery_app,但 app/__init__.py 里没 from .celery_app import celery_app
– celery_app.py 里直接写了 from app import db,而此时 app 尚未初始化,引发循环导入或 ImportError
任务函数里硬写 current_app 或 db 会立刻失败
以下写法全部不可行:
– @celery.task def send_mail(): user = User.query.get(1) # 崩
– @celery.task def send_mail(): print(current_app.config['SECRET_KEY']) # 崩
– @celery.task def send_mail(): db.session.rollback() # 崩
正确做法是把原始参数传进去,进任务后再重建上下文:
– 单文件结构:用 with app.app_context():,前提是任务模块能拿到 app 实例
– 工厂模式:在任务里调 create_app(),再 with app.app_context():
– 更稳妥:封装数据库操作为纯函数,自己管理 session 生命周期,结尾加 db.session.remove()
立即学习“Python免费学习笔记(深入)”;
ContextTask 封装是工厂模式下的推荐解法
如果使用 create_app(),不要手写 make_celery(),改用官方推荐的 celery.init_app(app),并配合自定义 ContextTask:
– 在 celery_app.py 中设置 celery.Task = ContextTask
– ContextTask.__call__ 里通过 self.app.flask_app.app_context() 推送上下文
– 关键前提是:你得在初始化 celery 时,把 app 绑定到 celery.app.flask_app 上(比如在 create_app() 里做这一步)
否则 self.app.flask_app 仍是 None,照样崩
最容易被忽略的一点:worker 进程里没有请求上下文(request)、也没有应用上下文(current_app),这不是 bug,是设计使然。所有上下文都得显式推送,且必须基于一个真实存在的 Flask 实例——这个实例不能靠“猜”或“全局变量”来获取,得从初始化路径里稳稳地传进来。


















