Is there any way to have custom validation logic in a FastAPI query parameter?
example
I have a FastAPI app with a bunch of request handlers taking Path components as query parameters. For example:
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)
I was wondering if there is way to ensure that the handler is not even called when the given file path is absolute.
Now I have to repeat myself with _raise_if_non_relative_path everywhere.
what I tried
fastapi.Query: pathlib.Path with validation logic in __init__: pathlib.PosixPath.@app.middleware: Path objects.class RelativePath(pydantic.Basemodel): path field, which I can validate however I want.
Unfortunately, this does not work for query parameters.
If I do this, the request handler insists on having a json content body.
Or at least that is what the swagger docs say.This is the kind of validation that the Depends dependency management function is well-suited for. It allows you to define dependencies for given view functions, and add logic to validate (and lazily create) those dependencies. This gives a set of composable dependencies that can be re-used in those views where they are required.
You can create a relative_where_query dependency, and then depend on that to perform any required validation:
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)
This gives small, easily readable (and understandable) view functions, while the dependency ("I need a relative path from the where query parameter") has been moved to its own definition.
You can then re-use this dependency in every view function that require a relative path from the where query parameter (and you can further decompose and recompose these dependencies further if necessary).
Update 2023-05-01: If you want to generalize this handling to decouple the field name from the validation function, you can do that by generating a dynamic dependency. This examples creates a dynamic base model with Pydantic to do it - but there's surely other ways.
Since we now raise the error inside a Pydantic validator, we add an exception handler to give a proper error response if the path isn't valid.
The magic itself happens inside path_validator - this function takes the name given to it, uses it as a field name in a create_model call - which dynamically creates a Pydantic model from its parameters. This model is then used as a dependency in the controller.
The name given as an argument to the function will be the actual parameter name - both when being called (i.e. in the URL in this case) and when the documentation is displayed. The path name in the controller function itself is only relevant inside the function, and isn't used in the documentation or as the query parameter (the field name in the model is used for that).
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)
You can now reuse this in any location you want by adding a dependency on a given parameter name:
def new_file(path: Path = Depends(path_validator("my_path"))):
...
def old_file(path: Path = Depends(path_validator("where"))):
...
Note: There is atm still a chance you may be using an older version of Pydantic and there is some chance it will require a modified solution. This is for projects using Pydantic >= 2.0.
FastAPI uses Pydantic under the hood also for validating parameters not defined as a Pydantic model, and it passes along the extra arguments to Path(), Query() and Body() to the FieldInfo object in Pydantic.
This means you can pass an extra annotation= argument to Query() and Path(), which can then be used to define a custom validator with AfterValidator().
from typing import Annotated
from fastapi import APIRouter, Path, Query
router = APIRouter("/prefix")
# Define validator according to Pydantic patterns
# https://docs.pydantic.dev/latest/concepts/validators/
def lowercase_validator(value: str) -> str:
assert value.lower() == value, f"{value} is not lowercase"
return value
# Define route using Path and Query parameters
# https://fastapi.tiangolo.com/tutorial/path-params-numeric-validations/
# https://fastapi.tiangolo.com/tutorial/query-params-str-validations/
@router.get("/{path}")
async def some_path(
path: str = Path(annotation=Annotated[str, AfterValidator(lowercase_validator)]),
arg: str = Query(annotation=Annotated[str, AfterValidator(lowercase_validator)]),
):
pass
This solution leads to FastAPI directly handling this as a part of the normal validation and returning the same type of 422 Unprocessable Entity error as all other validation failures.
In the case of the original question then you would make a validator something like
import pathlib
def validate_relative_path(path: pathlib.Path):
assert not path.is_absolute(), f"{path} is not relative"
return Path
And use it in the route something like
@app.get("/new",)
def new_file(where: pathlib.Path = Query(annotation=Annotated[str, AfterValidator(validate_relative_path)])):
pass
If you love us? You can donate to us via Paypal or buy me a coffee so we can maintain and grow! Thank you!
Donate Us With