Flask框架下原生SQLAlchemy会话的正确通用使用方案

你当前代码的核心问题是全局复用单个SQLAlchemy Session实例，并且手动在路由分支中调用close()的逻辑完全无法覆盖异常中断场景：只要视图函数在close()执行前抛出任何错误（参数校验失败、序列化报错、数据库操作异常），当前会话的残留事务、未释放连接就会污染后续请求，触发「上一会话未正常关闭、存在回滚错误」的报错。

SQLAlchemy的Session设计天生是单工作单元单实例，对应Web场景就是单请求单会话，所有会话的生命周期绑定请求周期，统一做初始化、提交/回滚、销毁，不需要在每个路由里重复写事务控制逻辑。

依靠Flask自带的请求生命周期钩子即可实现全API统一的会话管理，不需要在任何路由中单独写try-except或手动调用commit()/rollback()/close()，且100%覆盖异常中断场景。

第一步：修正会话初始化逻辑 #

不要提前创建全局会话实例，只保留线程安全的会话工厂：

from flask import Flask, g, request
from sqlalchemy.orm import sessionmaker

# 替换为你自己的引擎创建逻辑
engine = mysql_engine()
# 会话工厂，所有请求的会话都由这个工厂创建，本身是线程安全的
SessionFactory = sessionmaker(bind=engine, expire_on_commit=False)
app = Flask(__name__)

第二步：绑定请求级会话自动清理逻辑 #

利用Flask的before_request和teardown_request钩子，前者为每个请求创建独立会话，后者无论请求正常返回还是抛出异常都会执行，统一做事务收尾和连接释放：

@app.before_request
def inject_db_session():
    # 会话绑定到Flask g对象，每个请求的g上下文完全隔离，不存在线程安全问题
    g.db_session = SessionFactory()

@app.teardown_request
def auto_cleanup_session(exc):
    session = g.pop('db_session', None)
    if not session:
        return
    try:
        # 如果请求过程存在未捕获异常，直接回滚
        if exc:
            session.rollback()
        else:
            # 正常请求统一提交事务，只读查询的commit无额外副作用
            session.commit()
    except Exception:
        # 提交失败时强制回滚
        session.rollback()
        raise
    finally:
        # 无论成功失败，最终关闭会话，归还数据库连接到连接池
        session.close()

第三步：路由层直接使用 #

所有路由不需要再做任何会话手动管理，直接从g对象取会话操作即可：

@app.route('/get-score', methods=['POST'])
def handle_get_score():
    user_id = request.json.get('user_id')
    # 直接使用g.db_session查询，不需要手动关闭
    row = g.db_session.query(DB_Model_1) \
        .filter(DB_Model_1.user_id == user_id) \
        .first()
    if not row:
        return {}
    return {
        'userId': row.user_id,
        'score' : row.score
    }

@app.route('/insert-score', methods=['POST'])
def handle_insert_score():
    new_record = DB_Model_1(
        user_id=request.json.get('user_id'),
        score=request.json.get('score')
    )
    g.db_session.add(new_record)
    # 不需要手动commit，钩子会自动处理
    return {"code": 0, "msg": "success"}

# update、delete路由写法完全一致，直接操作g.db_session即可

• teardown_request是Flask框架保证必定执行的钩子，哪怕视图函数抛出500错误、中途被拦截器中断、返回响应前出现代码异常，都会进入清理逻辑，从根源避免会话残留。
• 不要使用全局单会话：Session本身不是线程安全的，并发场景下全局单会话不仅会出现事务污染，还会导致数据库连接被长期占用，最终耗尽连接池。
• 如果业务逻辑中需要手动回滚（比如校验不通过需要回滚已执行的写操作），直接在视图中调用g.db_session.rollback()即可，不会影响最终的自动清理流程。
• expire_on_commit=False参数避免提交后访问对象属性触发额外的数据库查询，适配Web场景的返回逻辑。

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