You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

FastAPI内部调用带会话校验装饰器的接口时,如何避免重复会话检查?

FastAPI内部调用带会话校验装饰器的接口时,如何避免重复会话检查?

我完全懂你遇到的麻烦——自己写的会话校验装饰器在接口内部调用其他带装饰的接口时,会重复触发两次会话检查,而你又不想把每个接口都拆成路由层和实现层,确实挺闹心的。下面给你几个实用又优雅的解决思路,你可以根据自己的代码风格挑:


方案一:给装饰器加「跳过校验」的开关(最小改动)

这种方法不用大改现有代码结构,只给require_session装饰器加一个可选参数,允许内部调用时手动跳过校验。

先修改你的装饰器:

from functools import wraps
import fastapi as fa
import exc  # 你的异常模块

def require_session(skip_check: bool = False):
    # 带参数的装饰器需要多一层嵌套,同时兼容无参数直接使用的情况
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            # 优先检查是否需要跳过校验
            if skip_check:
                return await func(*args, **kwargs)
            
            # 原有的会话校验逻辑
            request = kwargs.get('request')
            if not request:
                raise exc.access_denied()
                
            session_key = request.headers.get('X-Session-Key')
            if session_key and Session.check(session_key):
                return await func(*args, **kwargs)
            
            raise exc.access_denied()
        return wrapper

    # 兼容@require_session直接使用的场景(不带参数)
    if callable(skip_check):
        func = skip_check
        skip_check = False
        return decorator(func)
    return decorator

然后内部调用时,用带开关的版本包装原函数:

@administrators_router.post('')
@require_session
async def create_administrator(request: fa.Request, data: AdministratorSchema) -> AdministratorDBSchema:
    data_dump = data.model_dump()
    if isinstance(data_dump['employee'], int):
        # 内部调用时手动跳过校验
        await require_session(skip_check=True)(get_employee)(request, data_dump['employee'])
    try:
        admin = Administrator.create(**data_dump)
    except pw.IntegrityError as e:
        raise exc.integrity(e)
    return admin

方案二:用Request.state标记已校验(最省心)

利用FastAPI的request.state对象存储一个「会话已校验」的标记,装饰器先检查这个标记,已校验过就直接放行。

修改require_session装饰器:

def require_session(func):
    @wraps(func)
    async def wrapper(*args, **kwargs):
        request = kwargs.get('request')
        # 先检查请求状态里的校验标记
        if hasattr(request.state, 'session_verified') and request.state.session_verified:
            return await func(*args, **kwargs)
        
        # 原有的会话校验逻辑
        session_key = request.headers.get('X-Session-Key')
        if session_key and Session.check(session_key):
            # 标记会话已校验,后续调用直接跳过
            request.state.session_verified = True
            return await func(*args, **kwargs)
        
        raise exc.access_denied()
    return wrapper

这个方案几乎不需要改业务代码,只要装饰器逻辑加几行就行。当第一个带装饰器的接口被调用时,会在request.state上打标记,后续内部调用任何带require_session的接口时,都会自动跳过重复校验,完美适配你的现有代码!


方案三:改用FastAPI Depends依赖注入(最佳实践)

其实FastAPI的依赖注入系统就是为了解决这类重复校验问题而设计的,比自定义装饰器更灵活,也更符合框架的设计理念。

首先把会话校验写成一个依赖项:

async def verify_session(request: fa.Request):
    session_key = request.headers.get('X-Session-Key')
    if not session_key or not Session.check(session_key):
        raise exc.access_denied()
    # 可以返回会话对象,后续业务逻辑需要时直接用
    return Session.check(session_key)

然后在路由里用Depends代替装饰器:

# 员工路由
@employees_router.get('/{id}')
async def get_employee(
    request: fa.Request, 
    id: int,
    # 自动注入会话校验
    _=fa.Depends(verify_session)
) -> EmployeeDBSchema:
    employee = Employee.get_or_none(id=id)
    if employee is None:
        raise exc.instance_by_field_not_found(Employee, id)
    return employee

# 管理员路由
@administrators_router.post('')
async def create_administrator(
    request: fa.Request, 
    data: AdministratorSchema,
    _=fa.Depends(verify_session)
) -> AdministratorDBSchema:
    data_dump = data.model_dump()
    if isinstance(data_dump['employee'], int):
        # 内部调用时直接传参,不需要处理校验(当前路由已经做过了)
        await get_employee(request, data_dump['employee'], _=None)
    try:
        admin = Administrator.create(**data_dump)
    except pw.IntegrityError as e:
        raise exc.integrity(e)
    return admin

这种方法的好处是后续扩展更方便——比如要在业务逻辑里用会话信息,直接把依赖返回的会话对象作为参数就行,不用再从请求里取。


备注:内容来源于stack exchange,提问作者kos1posha

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.04.15 09:34:36