Блог

Я хочу добавить 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.routes объекте APIRouter . Второй способ не работает, потому что он не может проверить pydantic object. В нем говорится о любой модели запроса missing args, missing kwargs .

Итак, мой вопрос: как я могу добавить какие-либо декораторы к конечным точкам FastAPI? Должен ли я войти router.routes и изменить существующую конечную точку? Или использовать некоторые functools.wraps подобные функции?

Как я могу добавить какие-либо декораторы к конечным точкам 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(...) это тоже декоратор.

Вот как вы можете использовать декоратор, который добавляет дополнительные параметры в обработчик маршрута:

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

Просто используйте зависимости внутри декоратора операции path:

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