如何将自定义装饰器添加到 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 创建了一个路由器，该路由器保存到 APIRouter 对象的 self.routes 中。第二种方法不起作用，因为它无法验证 pydantic 对象。对于任何请求模型，它表示 missing args, missing kwargs.

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

Answer 1

How can I add any decorators to FastAPI endpoints?

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

<b>from functools import wraps</b>

from fastapi import FastAPI
from pydantic import BaseModel


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


app = FastAPI()


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

    return wrapper</b>


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

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

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

<b>from fastapi import Request</b>


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

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

Answer 2

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

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!"}

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

How to add a custom decorator to a FastAPI route?

python

python-decorators

fastapi

pydantic