Celery任务执行db.session.commit时报‘Outside of request context’错误求助

问题详情 #

在使用Celery异步修改SQLAlchemy数据表时，任务能正常查询并打印目标记录，但执行db.session.commit()时出现以下警告：

[2023-12-05 11:23:21,575: WARNING/ForkPoolWorker-6] Working outside of request context.
This typically means that you attempted to use functionality that needed an active HTTP request. Consult the documentation on testing for information about how to avoid this problem.

相关代码片段 #

Celery任务代码

@celery.task
def background_job(id, request_text):  
    app = current_app._get_current_object()
    with app.app_context():
        try:  
            row = Request.query.filter_by(id=id).first()
            row.request_text = request_text
            print(row)              # 正常执行：打印更新后的记录
            db.session.commit()     # 执行失败：触发上述警告
        except Exception as error:
            print("An error occurred:", error)
    return 

Celery Worker脚本

#!/bin/env python
import os
from app import celery, create_app

app = create_app(os.getenv('FLASK_CONFIG') or 'default')
app.app_context().push()

主应用代码

from flask import Flask
from config import Config
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from flask_socketio import SocketIO
from celery import Celery

socketio = SocketIO()
db = SQLAlchemy()
migrate = Migrate()

celery = Celery(    __name__, 
                    broker=Config.broker_url, 
                    include=Config.include, 
                    broker_connection_retry_on_startup = Config.broker_connection_retry_on_startup)


def create_app(debug=False):
    """Create an application."""
    app = Flask(__name__)
    app.config.from_object(Config)
    celery.conf.update(app.config)
    db.init_app(app)
    migrate.init_app(app, db)

    # API Endpoints
    from .endpoints_rest import endpoints_rest as blueprint_endpoints_rest
    app.register_blueprint(blueprint_endpoints_rest)
    from .endpoints_socketio import endpoints_socketio as blueprint_endpoints_socketio
    app.register_blueprint(blueprint_endpoints_socketio) 

    socketio.init_app(app)

    return app

解决思路与方案 #

1. 理解警告根源 #

Flask-SQLAlchemy的db.session是请求作用域的scoped session，默认绑定到Flask的请求上下文。Celery任务运行在独立的Worker进程中，没有HTTP请求上下文，直接使用全局db.session会触发该警告。

2. 调整任务中的数据库会话管理 #

修改Celery任务代码，手动管理数据库会话，避免依赖请求上下文：

@celery.task
def background_job(id, request_text):  
    try:  
        # 手动创建并管理会话
        session = db.session()
        row = session.query(Request).filter_by(id=id).first()
        if row:
            row.request_text = request_text
            session.commit()
            print(row)
        else:
            session.rollback()
    except Exception as error:
        print("An error occurred:", error)
        if 'session' in locals():
            session.rollback()
    finally:
        if 'session' in locals():
            session.remove()
    return 

3. 简化Worker上下文配置 #

Worker脚本已经push了app上下文，任务内无需再通过current_app获取并嵌套app.app_context()，可以直接使用全局的db对象。

4. 可选：配置SQLAlchemy会话作用域 #

如果希望在Celery中更顺畅地使用Flask-SQLAlchemy，可以在Flask配置类中调整会话作用域为进程作用域（需注意并发安全）：

# 在Config类中添加
SQLALCHEMY_SCOPE_SESSION = 'process'

不过更推荐的方式还是在Celery任务中手动管理会话，避免作用域冲突。

5. 验证修改 #

重启Celery Worker和Flask应用，触发异步任务后检查：

• 数据表是否成功更新
• 控制台是否不再出现"Working outside of request context"警告

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