有没有办法在 FastAPI 查询参数中拥有自定义验证逻辑?
例子
我有一个FastAPI应用程序,其中有一堆请求处理程序,将Path组件作为查询参数。例如:
def _raise_if_non_relative_path(path: Path):
if path.is_absolute():
raise HTTPException(
status_code=409,
detail=f"Absolute paths are not allowed, {path} is absolute."
)
@app.get("/new",)
def new_file(where: Path):
_raise_if_non_relative_path(where)
# do save a file
return Response(status_code=requests.codes.ok)
@app.get("/new",)
def delete_file(where: Path):
_raise_if_non_relative_path(where)
# do save a file
return Response(status_code=requests.codes.ok)
Run Code Online (Sandbox Code Playgroud)
我想知道是否有办法确保当给定的文件路径是绝对路径时甚至不会调用处理程序。现在我必须到处重复自己_raise_if_non_relative_path。
我尝试过什么
fastapi.Query:pathlib.Path带有验证逻辑的子类__init__:pathlib.PosixPath.@app.middleware:Path对象。class RelativePath(pydantic.Basemodel):path字段的类,我可以根据需要对其进行验证。不幸的是,这不适用于查询参数。如果我这样做,请求处理程序会坚持使用 json 内容主体。或者至少 swagger 文档是这么说的。依赖Depends管理功能非常适合这种验证。它允许您定义给定视图函数的依赖关系,并添加逻辑来验证(并延迟创建)这些依赖关系。这提供了一组可组合的依赖项,可以在需要的视图中重用它们。
您可以创建relative_where_query依赖项,然后依赖它来执行任何所需的验证:
from fastapi import Depends, FastAPI, Response, Query
from fastapi.exceptions import HTTPException
from pathlib import Path
import requests
app = FastAPI()
def relative_where_query(where: Path = Query(...)):
if where.is_absolute():
raise HTTPException(
status_code=409,
detail=f"Absolute paths are not allowed, {where} is absolute."
)
return where
@app.get("/new")
def new_file(where: Path = Depends(relative_where_query)):
return Response(status_code=requests.codes.ok)
Run Code Online (Sandbox Code Playgroud)
这提供了小型、易于阅读(且易于理解)的视图函数,而依赖项(“我需要查询参数的相对路径where”)已移至其自己的定义。
然后,您可以在每个需要查询参数的相对路径的视图函数中重复使用此依赖项where(如果需要,您可以进一步分解和重新组合这些依赖项)。
更新 2023-05-01:如果您想推广此处理以将字段名称与验证函数分离,您可以通过生成动态依赖项来实现。本示例使用 Pydantic 创建一个动态基础模型来完成此任务 - 但肯定还有其他方法。
由于我们现在在 Pydantic 验证器内引发错误,因此我们添加了一个异常处理程序,以便在路径无效时给出正确的错误响应。
魔法本身就发生在内部path_validator- 该函数采用给定的名称,将其用作调用中的字段名称create_model- 它根据其参数动态创建 Pydantic 模型。然后将该模型用作控制器中的依赖项。
作为函数参数给出的名称将是实际参数名称 - 无论是在调用时(即本例中的 URL)还是显示文档时。控制器函数本身中的名称path仅在函数内部相关,并且不在文档中使用或用作查询参数(模型中的字段名称用于此目的)。
from fastapi import Depends, FastAPI, Response, Query, Request
from fastapi.exceptions import HTTPException
from fastapi.responses import JSONResponse
from pathlib import Path
from pydantic import create_model, validator
import requests
app = FastAPI()
class PathNotAbsoluteError(Exception):
pass
@app.exception_handler(PathNotAbsoluteError)
async def value_error_exception_handler(request: Request, exc: PathNotAbsoluteError):
return JSONResponse(
status_code=400,
content={"message": str(exc)},
)
def path_is_absolute(cls, value):
if not value.is_absolute():
raise PathNotAbsoluteError("Given path is not absolute")
return value
def path_validator(param_name):
validators = {
'pathname_validator': validator(param_name)(path_is_absolute)
}
return create_model(
"AbsolutePathQuery",
**{param_name: (Path, Query(...))},
__validators__=validators,
)
@app.get("/path")
def new_file(path: Path = Depends(path_validator("where"))):
return Response(status_code=requests.codes.ok)
Run Code Online (Sandbox Code Playgroud)
现在,您可以通过添加对给定参数名称的依赖关系在任何位置重用它:
def new_file(path: Path = Depends(path_validator("my_path"))):
...
def old_file(path: Path = Depends(path_validator("where"))):
...
Run Code Online (Sandbox Code Playgroud)