使用FastAPI带Security的依赖时如何避免创建多会话？

问题现象 #

按照SQLModel教程实现同步SQLite会话依赖：

def get_session():
    with Session(engine) as session:
        yield session

当路由同时依赖调用get_session的Dependency和Security时，会创建两个独立会话，触发SQLAlchemy错误：

sqlalchemy.exc.InvalidRequestError: Object '<User at 0x7f218cbf6a00>' is already attached to session '1' (this is '2')

简化代码示例：

from fastapi import Depends, Security
from fastapi.security import SecurityScopes
from sqlmodel import Session, create_engine, select


def get_session():
    with Session(engine) as session:
        yield session


async def get_current_user(
    security_scopes: SecurityScopes,
    token: str = Depends(oauth2_scheme),
    session: Session = Depends(get_session)
):
    # 省略Token解析逻辑
    user = session.get(User, token_data.user_uuid)
    if user is None:
        raise credentials_exception
    return user


async def get_current_app_user(
    current_user: User = Security(get_current_user, scopes=["app"])
):
    return current_user


@app.patch("/me", response_model=UserRead)
def update_current_user(
    user: UserUpdate,
    current_user: User = Depends(get_current_app_user),
    session: Session = Depends(get_session),
):
    user_data = user.dict(exclude_unset=True)
    for key, value in user_data.items():
        setattr(current_user, key, value)
    # 当前session与current_user绑定的session不是同一个
    session.add(current_user) 
    session.commit()
    session.refresh(current_user)
    return current_user

问题原因 #

当Security的scopes为非空列表时，FastAPI会将scopes纳入依赖缓存键（cache_key），导致即使use_cache=True，也无法复用get_session依赖，最终创建两个会话：

self.cache_key = (self.call, tuple(sorted(set(self.security_scopes or []))))

解决方案 #

方案1：从用户对象复用会话 #

既然current_user已经绑定了会话，直接从对象中获取会话即可，无需重复注入新的session依赖：

@app.patch("/me", response_model=UserRead)
def update_current_user(
    user: UserUpdate,
    current_user: User = Depends(get_current_app_user),
):
    user_data = user.dict(exclude_unset=True)
    for key, value in user_data.items():
        setattr(current_user, key, value)
    # 从current_user获取绑定的会话
    session = Session.object_session(current_user)
    session.add(current_user) 
    session.commit()
    session.refresh(current_user)
    return current_user

方案2：实现请求级单例会话 #

修改get_session，通过FastAPI请求状态存储会话，确保单个请求内仅创建一次会话：

from fastapi import Request

def get_session(request: Request):
    if "session" not in request.state:
        request.state.session = Session(engine)
    yield request.state.session
    # 请求结束后关闭会话，避免连接泄漏
    request.state.session.close()

此方式下，无论依赖被调用多少次，同一请求内只会复用同一个会话。

方案3：自定义Security缓存键（谨慎使用） #

如果必须保留两次依赖注入，可以自定义Security子类，修改缓存键生成逻辑，忽略scopes的影响：

from fastapi import Security as FastAPISecurity

class Security(FastAPISecurity):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        # 重写cache_key，仅保留依赖函数，忽略scopes
        self.cache_key = (self.call,)

替换原生Security使用：

async def get_current_app_user(
    current_user: User = Security(get_current_user, scopes=["app"])
):
    return current_user

⚠️ 注意：此方式会让不同scopes的Security依赖共享缓存结果，仅适用于scopes不影响get_current_user返回结果的场景。

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