π‘οΈ Adding Validation to Query Parameters
So far, we have validated:
β Request bodies using Pydantic
β Path parameters using
Path()
Now we will validate:
π Query parameters
Just like path parameters, query parameters can be validated using FastAPIβs Query class.
This ensures that only acceptable values reach your business logic.
π§ Step 1: Import Query
Begin by importing Query from FastAPI.
from typing import Optional
from fastapi import FastAPI, Path, Query π
from pydantic import BaseModel, Fieldπ Query allows you to apply constraints directly inside your endpoint signature.
This means validation happens:
Before the function executes
Automatically
Without manual checks
π§ͺ Step 2: Validate the Book Rating Query Parameter
Letβs update our rating filter endpoint.
@app.get("/books/") # π Query parameter
async def read_book_by_rating(book_rating: int = Query(gt=0, lt=6)):
books_to_return = []
for book in BOOKS:
if book.rating == book_rating:
books_to_return.append(book)
return books_to_returnπ What Does Query(gt=0, lt=6) Do?
It enforces:
book_ratingmust be greater than 0book_ratingmust be less than 6
π This effectively limits valid ratings to:
1, 2, 3, 4, 5If someone sends:
0 OR 10
FastAPI will automatically return:
HTTP 422 β Unprocessable EntityAnd your function will never run.
π§ Backend Fundamental Insight
Notice something important:
We did NOT write:
if book_rating < 1 or book_rating > 5:Why?
Because:
π Validation belongs at the boundary, not inside business logic.
This keeps your code:
Cleaner
More maintainable
More scalable
π Step 3: Test in the Browser
Try the following:
Enter
0Enter
10
You should see:
422 β Unprocessable EntityNow try:
5You should receive matching books.
π Step 4: Validate the Published Date Query Parameter
We already enforce rules for published_date in BookRequest:
published_date: int = Field(gt=1999, lt=2031)Letβs apply the same rules to the query endpoint.
@app.get("/books/publish/")
async def read_books_by_publish_date(
published_date: int = Query(gt=1999, lt=2031)):
books_to_return = []
for book in BOOKS:
if book.published_date == published_date:
books_to_return.append(book)
return books_to_returnπ What This Enforces
Year must be greater than 1999
Year must be less than 2031
Valid range: 2000β2030
If someone submits:
1990 OR 2050
FastAPI automatically responds with:
HTTP 422 β Unprocessable Entityπ Step 5: Test in the Browser
Try:
A valid year like
2025An invalid year like
1980
Observe how validation blocks bad input before logic runs.
ποΈ books2.py
from typing import Optional
from fastapi import FastAPI, Path, Query
from pydantic import BaseModel, Field
app = FastAPI()
class Book:
id: int
title: str
author: str
description: str
rating: int
published_date: int
def __init__(self, id, title, author, description, rating, published_date):
self.id = id
self.title = title
self.author = author
self.description = description
self.rating = rating
self.published_date = published_date
class BookRequest(BaseModel):
id: Optional[int] = Field(description="ID is not required", default=None)
title: str = Field(min_length=3)
author: str = Field(min_length=1)
description: str = Field(min_length=1, max_length=100)
rating: int = Field(gt=1, lt=6)
published_date: int = Field(gt=1999, lt=2031)
model_config = {
"json_schema_extra": {
"example": {
"title": "A new book",
"author": "Name of the author",
"description": "The description of the book",
"rating": 5,
"published_date": 2029
}
}
}
BOOKS = [
Book(1, "Computer Science Pro", "codingwithrobby", "A very nice book", 3, 2030),
Book(2, "Be Fast with FastAPI", "codingwithrobby", "A great book", 5, 2030),
Book(3, "Master Endpoints", "codingwithrobby", "An awesome book", 5, 2029),
Book(4, "HP1", "Author 1", "Book Description", 2, 2028),
Book(5, "HP2", "Author 2", "Book Description", 3, 2027),
Book(6, "HP3", "Author 3", "Book Description", 1, 2026)
]
@app.get("/books")
async def read_all_books():
return BOOKS
@app.post("/create_book")
async def create_book(book_request: BookRequest):
new_book = Book(**book_request.model_dump())
BOOKS.append(find_book_id(new_book))
@app.get("/books/{book_id}")
async def read_book(book_id: int):
for book in BOOKS:
if book.id == book_id:
return book
@app.get("/books/")
async def read_book_by_rating(book_rating: int = Query(gt=0, lt=6)):
books_to_return = []
for book in BOOKS:
if book.rating == book_rating:
books_to_return.append(book)
return books_to_return
@app.put("/books/update_book")
async def update_book(book: BookRequest):
for i in range(len(BOOKS)):
if BOOKS[i].id == book.id:
BOOKS[i] = book
@app.delete("/books/{book_id}")
async def delete_book(book_id: int):
for i in range(len(BOOKS)):
if BOOKS[i].id == book_id:
BOOKS.pop(i)
break
@app.get("/books/publish/")
async def read_books_by_publish_date(
published_date: int = Query(gt=1999, lt=2031)):
books_to_return = []
for book in BOOKS:
if book.published_date == published_date:
books_to_return.append(book)
return books_to_return
def find_book_id(book: Book):
book.id = 1 if len(BOOKS) == 0 else BOOKS[-1].id + 1
return book