问题描述
我想向端点添加一个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之类的函数？
解决方法
如何将任何装饰器添加到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!"}