如何向 FastAPI 路由添加自定义装饰器？

Question

我想向auth_required我的端点添加一个装饰器。（请考虑这个问题是关于装饰器，而不是中间件）

所以一个简单的装饰器看起来像这样：

def auth_required(func):
    def wrapper(*args, **kwargs):
        if user_ctx.get() is None:
            raise HTTPException(...)
        return func(*args, **kwargs)
    return wrapper

所以有2种用法：

@auth_required
@router.post(...)

或者

@router.post(...)
@auth_required

第一种方法不起作用，因为router.post创建了一个保存到self.routesAPIRouter 对象中的路由器。第二种方法不起作用，因为它无法验证 pydantic 对象。对于任何请求模型，它说missing args, missing kwargs。

所以我的问题是 - 如何向 FastAPI 端点添加任何装饰器？我应该进入router.routes并修改现有端点吗？或者使用一些functools.wraps类似的功能？

Answer 1

如何向 FastAPI 端点添加任何装饰器？

正如你所说，你需要使用@functools.wraps(...)--(PyDoc)装饰器，

from functools import wraps

from fastapi import FastAPI
from pydantic import BaseModel


class SampleModel(BaseModel):
    name: str
    age: int


app = FastAPI()


def auth_required(func):
    @wraps(func)
    async def wrapper(*args, **kwargs):
        return await func(*args, **kwargs)

    return wrapper


@app.post("/")
@auth_required # Custom decorator
async def root(payload: SampleModel):
    return {"message": "Hello World", "payload": payload}

此方法的主要警告是您无法访问request包装器中的对象，我认为这是您的主要意图。

如果您需要访问请求，则必须将参数添加到路由器函数中，

from fastapi import Request


@app.post("/")
@auth_required  # Custom decorator
async def root(request: Request, payload: SampleModel):
    return {"message": "Hello World", "payload": payload}

我不确定FastAPI中间件有什么问题，毕竟@app.middleware(...)它也是一个装饰器。

Answer 2

只需使用路径操作装饰器内部的依赖项：

from fastapi import Depends, FastAPI, Header, HTTPException

app = FastAPI()


async def verify_token(x_token: str = Header()):
    if x_token != "fake-super-secret-token":
        raise HTTPException(status_code=400, detail="X-Token header invalid")


async def verify_key(x_key: str = Header()):
    if x_key != "fake-super-secret-key":
        raise HTTPException(status_code=400, detail="X-Key header invalid")
    return x_key


@app.get("/items/", dependencies=[Depends(verify_token), Depends(verify_key)])
async def read_items():
    return [{"item": "Foo"}, {"item": "Bar"}]

Answer 3

以下是如何使用装饰器向路由处理程序添加额外参数：

from fastapi import FastAPI, Request
from pydantic import BaseModel


class SampleModel(BaseModel):
    name: str
    age: int


app = FastAPI()

def do_something_with_request_object(request: Request):
    print(request)

def auth_required(handler):
    async def wrapper(request: Request, *args, **kwargs):
        do_something_with_request_object(request)
        return await handler(*args, **kwargs)

    # Fix signature of wrapper
    import inspect
    wrapper.__signature__ = inspect.Signature(
        parameters = [
            # Use all parameters from handler
            *inspect.signature(handler).parameters.values(),

            # Skip *args and **kwargs from wrapper parameters:
            *filter(
                lambda p: p.kind not in (inspect.Parameter.VAR_POSITIONAL, inspect.Parameter.VAR_KEYWORD),
                inspect.signature(wrapper).parameters.values()
            )
        ],
        return_annotation = inspect.signature(handler).return_annotation,
    )

    return wrapper


@app.post("/")
@auth_required # Custom decorator
async def root(payload: SampleModel):
    return {"message": f"Hello {payload.name}, {payload.age} years old!"}