有没有办法在 FastAPI 中使用装饰器进行请求 header 检查？

Question

我目前正在查看 the FastAPI docs 中提供的检查传入请求 header 的示例。

我想知道是否有一种方法可以在路由上使用装饰器来实现 header 检查，而不是在每个端点函数中重复检查代码？类似于：

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

您可以创建一个APIRouter对象，而不是直接引用app对象，并将依赖项添加为整个路由器的依赖项：

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)

有没有办法在 FastAPI 中使用装饰器进行请求 header 检查？

Is there a way to do the request header check using a decorator in FastAPI?

python

http

decorator

request

fastapi