如何为 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

如何向 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 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!"}

Answer 3

只需在路径操作装饰器中使用dependencies：

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 4

除了 JPG 的答案，您还可以使用

Request

访问装饰器中的

kwargs.get('request')

对象。一个完整的装饰器看起来像：

def render_template(template):
    """decorator to render a template with a context"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):

            # access request object
            request = kwargs.get('request')

            context = func(*args, **kwargs)
            if context is None:
                context = {}
            return templates.TemplateResponse(template, {**context, 'request': request})
        return wrapper
    return decorator

装饰函数需要将

Request

作为参数。