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

FastAPI中是否可以使用装饰器实现请求头校验功能?

解决方案

你现有代码的核心问题是装饰器的wrapper函数只接收了request参数,没有透传端点函数的其他位置参数、关键字参数,同时FastAPI原生提供了更简洁的依赖注入方式来实现统一请求头校验,不需要手动写装饰器。

方案1:修正你的自定义装饰器

修改后的装饰器代码如下:

from functools import wraps
from fastapi import Request, HTTPException

SECRET_KEY = "你的实际密钥" # 请替换为真实业务密钥

def check_header_api_key(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        # 优先从关键字参数中取request对象
        request = kwargs.get("request")
        if not request:
            # 遍历位置参数匹配Request类型对象
            for arg in args:
                if isinstance(arg, Request):
                    request = arg
                    break
        if not request or request.headers.get("SECRET") != SECRET_KEY:
            raise HTTPException(status_code=401, detail="Invalid client secret")
        # 透传所有参数给原端点函数
        return func(*args, **kwargs)
    return wrapper

对应的端点函数不需要修改,保持你原有的写法即可正常运行。

方案2:更符合FastAPI最佳实践的依赖注入实现

FastAPI的依赖注入特性天生支持全局、路由组、单端点级别的统一校验,比自定义装饰器更简洁,也能完美兼容FastAPI的参数解析、OpenAPI文档生成:

  1. 首先定义校验依赖
from fastapi import Depends, HTTPException, status
from fastapi.security import APIKeyHeader

# 定义要校验的请求头字段
api_key_header = APIKeyHeader(name="SECRET", auto_error=False)

async def verify_api_key(api_key: str = Depends(api_key_header)):
    if api_key != SECRET_KEY:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid client secret"
        )
  1. 单个端点使用时,直接在端点参数中加依赖即可,不需要手动处理request对象:
@app.post("/course", response_model=schemas.Course, status_code=200)
def create_course(
    course: schemas.CourseCreate, 
    db: Session = Depends(get_db),
    _: None = Depends(verify_api_key)
):
    # 原有业务逻辑完全不变
    db_course = crud.get_course(db, course=course)
    if db_course:
        raise HTTPException(status_code=400, detail="This course has already been created.")
    return crud.create_course(db=db, course=course)
  1. 如果需要给某一组路由统一加校验,不需要每个端点都加依赖,可以用APIRouter的dependencies参数批量配置:
from fastapi import APIRouter

# 创建路由组时统一添加校验依赖
course_router = APIRouter(
    prefix="/course",
    dependencies=[Depends(verify_api_key)]
)

# 该路由组下的所有端点自动继承校验逻辑,无需额外配置
@course_router.post("/", response_model=schemas.Course, status_code=200)
def create_course(course: schemas.CourseCreate, db: Session = Depends(get_db)):
    # 业务逻辑保持不变
    ...

内容的提问来源于stack exchange,提问作者Daniel Qiao

相关产品推荐
方舟 Agent Plan

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

最近更新时间:2026.10.02 09:12:04