如何在FastAPI应用中隐式处理SQLAlchemy会话生命周期？

1. 依赖注入结合上下文管理器 #

利用FastAPI原生的依赖系统，定义一个会话依赖，通过上下文管理器自动处理会话的创建、提交/回滚和关闭逻辑，确保每个请求（含后台任务）绑定独立会话，且自动完成生命周期管理。

示例代码：

from sqlalchemy.orm import Session
from fastapi import Depends
from your_module import default_db

def get_db_session():
    session = default_db.create_session()
    try:
        yield session
        # 请求正常结束时自动提交，模拟Django模式
        session.commit()
    except Exception:
        session.rollback()
        raise
    finally:
        session.close()

# API视图中直接注入使用
@app.get("/items/")
def read_items(db: Session = Depends(get_db_session)):
    items = db.query(Item).all()
    return items

# 后台任务中同样支持注入
from fastapi import BackgroundTasks

@app.post("/create-item/")
def trigger_async_task(background_tasks: BackgroundTasks):
    background_tasks.add_task(process_item_async)
    return {"message": "Async task started"}

def process_item_async(db: Session = Depends(get_db_session)):
    new_item = Item(name="Async Item")
    db.add(new_item)
    db.commit()

2. 自定义装饰器封装函数 #

编写一个装饰器，在目标函数执行前初始化会话，执行后处理提交/回滚和关闭，可灵活应用于API视图、FastStream任务或后台任务。

示例代码：

from functools import wraps
import inspect
from your_module import default_db

def manage_session(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        session = default_db.create_session()
        try:
            # 自动为函数注入db参数（如果函数声明了该参数）
            if "db" in inspect.signature(func).parameters:
                kwargs["db"] = session
            result = func(*args, **kwargs)
            session.commit()
            return result
        except Exception:
            session.rollback()
            raise
        finally:
            session.close()
    return wrapper

# 应用到API视图
@app.get("/items/")
@manage_session
def read_items(db):
    items = db.query(Item).all()
    return items

# 应用到FastStream任务
@manage_session
@broker.subscriber("task_queue")
async def handle_stream_task(db, message):
    task = Task(status="processed", content=message)
    db.add(task)

3. 上下文变量（ContextVars）实现隐式会话获取 #

借助Python的contextvars模块，将会话绑定到当前执行上下文（请求/任务上下文），实现无需显式传递的隐式会话获取，同时自动隔离不同请求/任务的会话实例。

示例代码：

import contextvars
from functools import wraps
from sqlalchemy.orm import Session
from your_module import default_db

# 定义上下文变量存储会话
db_session_var = contextvars.ContextVar("db_session")

def get_db_session():
    try:
        return db_session_var.get()
    except LookupError:
        # 首次获取时创建会话并绑定到上下文
        session = default_db.create_session()
        db_session_var.set(session)
        return session

# 装饰器用于管理上下文会话的生命周期
def manage_context_session(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        session = default_db.create_session()
        token = db_session_var.set(session)
        try:
            result = func(*args, **kwargs)
            session.commit()
            return result
        except Exception:
            session.rollback()
            raise
        finally:
            db_session_var.reset(token)
            session.close()
    return wrapper

# 视图中隐式获取会话
@app.get("/items/")
@manage_context_session
def read_items():
    db = get_db_session()
    items = db.query(Item).all()
    return items

方案对比 #

• 依赖注入：FastAPI原生支持，代码简洁易维护，对后台任务适配性好，是最推荐的常规方案。
• 自定义装饰器：灵活度高，可快速适配现有函数，无需修改函数签名（若通过参数注入）。
• 上下文变量：实现完全隐式的会话调用，代码侵入性最低，但需要额外的上下文管理逻辑，适合追求极致简洁的场景。

以上方案均支持FastStream任务和FastAPI后台任务，能确保每个任务拥有独立会话，并在任务结束后自动完成清理。

内容的提问来源于stack exchange，提问作者Альберт Александров