是否有一种方法可以使用FastAPI中的装饰器来检查请求头?

Question

我目前正在查看FastAPI文档中提供的检查传入请求头的示例。

我想知道是否有一种方法可以在路由上使用装饰器来实现头检查，而不是在每个端点函数中重复检查代码?比如:

def check_header_api_key(func, *args, **kwargs):
@wraps(func)
def wrapper(request: Request):
if request.headers["SECRET"] != SECRET_KEY:
raise HTTPException(status_code=401, detail="Invalid client secret")
return func(request, *args, **kwargs)
return wrapper

@app.post("/course", response_model=schemas.Course, status_code=200)
@check_header_api_key
def create_course(course: schemas.CourseCreate, request: Request, db: Session = Depends(get_db)):
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)

我很熟悉我应该如何处理参数进入装饰器，有人可以帮助吗?

Answer 1

不直接引用app对象，你可以创建一个APIRouter对象，并将其作为整个路由器的依赖项:

authenticated = APIRouter(dependencies=[Depends(get_authenticated_user)])

你可以给路由器添加端点:

@authenticated.post('/course', ....)

你可以按照你的意愿组合这些路由器对象，要么直接在你的app下，要么作为彼此的子路由器:

app.include_router(authenticated)
# or with several sub routers:
authenticated_router = APIRouter(dependencies=[Depends(get_authenticated_user)])
public_router = APIRouter()
authenticated_router.include_router(courses.authenticated)
authenticated_router.include_router(users.authenticated)
public_router.include_router(users.public)
app.include_router(authenticated_router)
app.include_router(public_router)

这样你就可以根据需要组合和移动你的路由器，使它们需要或不需要身份验证，你可以进一步扩展它们到admin/private/etc。

Answer 2

解决了同样的问题。FastAPI将request处理程序传递给kwargs。如果您需要解析它，您的代码应该如下所示:

from fastapi import FastAPI, Request
from fastapi import HTTPException
def check_header_api_key(func):
@wraps(func)
def wrapper(request, *args, **kwargs):
if request.headers.get("SECRET", None) != SECRET_KEY:
raise HTTPException(status_code=401, detail="Invalid client secret")
return func(*args, **kwargs)
return wrapper

@app.post("/course", response_model=schemas.Course, status_code=200)
@check_header_api_key
def create_course(request: Request, course: schemas.CourseCreate, db: Session = Depends(get_db)):
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)