Path Parameter Data Validation | Query Parameter Data Validation
Path Parameter Data Validation
Here we will try to put validation on each API endpoint we have created in FastAPI.
Parts which we will try to validate is:
- If there is no book then are we returning any message as confirmation of validation?
- return desired message if your logic not executed as final return statement to validate your API endpoint run in place of showing null in output box or no message at output.
- How to validate input parameter of the method?
- We need to import path from FastAPI to implement it at the time of method definition to provide input constraints to be checked.
@app.get("/books/{book_id}")
async def read_book_by_id(book_id: int = Path(gt=0)):
# it will make sure that index will start from 1.
for book in BOOKS:
if book.id == book_id:
return book
return {"message": "book not found"}
Code:
from fastapi import FastAPI, Body, Path
from books3 import Book
from datavalidatorhelper import BookRequest
app = FastAPI()
BOOKS = [
Book(id=1, title="DSA with Java",
author="Puneet Kumar Singh",
description="A comprehensive guide to Data Structures
and Algorithms using Java.",
rating=5,
published_date=2014),
Book(id=2, title="Python Programming",
author="Puneet Kumar Singh",
description="An introduction to Python programming for beginners.",
rating=4,
published_date=2015),
Book(id=3, title="Machine Learning Basics",
author="Puneet Kumar Singh",
description="Understanding the fundamentals of Machine Learning.",
rating=4,
published_date=2014),
Book(id=4, title="Web Development with FastAPI",
author="Ramesh Jha",
description="Building web applications using FastAPI.",
rating=3,
published_date=2016),
Book(id=5, title="Data Science with Python",
author="Kalidas",
description="A guide to Data Science using Python.",
rating=2,
published_date=2018),
Book(id=6, title="Advanced Java Programming",
author="Kalpana Gupta",
description="Deep dive into advanced concepts of Java programming.",
rating=1,
published_date=2014)
]
# API endpoint to get all books detail unfiltered.
@app.get("/books") # static path to get all books
async def read_all_books():
"""
This function returns the list of all books.
"""
return BOOKS
""" post request without any input validation
@app.post("/create_book")
async def create_book(book_request=Body()):
BOOKS.append(book_request)
"""
#API endpoint to fetch book by using published_date as parameter
@app.get("/books/published/")
async def read_book_by_published_date(published_date: int):
books_to_return = []
for book in BOOKS:
if book.published_date == published_date:
books_to_return.append(book)
return books_to_return
# API endpoint to fetch book by using id
@app.get("/books/{book_id}")
async def read_book_by_id(book_id: int = Path(gt=0)):
# it will make sure that index will start from 1.
for book in BOOKS:
if book.id == book_id:
return book
return {"message": "book not found"}
# API endpoint to fetch books by rating
@app.get("/books/")
async def read_book_by_rating(rating: int):
book_to_return = []
for book in BOOKS:
if book.rating == rating:
book_to_return.append(book)
return book_to_return
#post request with input validator
@app.post("/creat_book")
async def create_book(book_request: BookRequest):
new_book = Book(**book_request.model_dump())
# new_book = Book(**book_request.dict()) -> if you are using
pydantic version 1 else the model_dump() for pydentic version 2.
# BOOKS.append(new_book) -> it will store the valid
input dump but you can have duplicate ids.
BOOKS.append(find_book_id(new_book))
# It will add latest id sequence number to id automatically and
will overwrite the provided input so you don't need to bother
about the id value sequence.
def find_book_id(book: Book):
if len(BOOKS) > 0:
book.id = BOOKS[-1].id + 1
else:
book.id = 1
return book
# API endpoint to update existing book entry
@app.put("/books/update_book")
async def update_book_by_id(book_request: BookRequest):
new_book=Book(**book_request.model_dump())
for i in range(len(BOOKS)):
if BOOKS[i].id == new_book.id:
BOOKS[i]=new_book
return {"message": "Book is updated", "detail": new_book}
return {"message": "Book id not found. Hence can't
add new book to library. Do use create book API endpoint to
add new book into library."}
# API endpoint to delete book entry from library
@app.delete("/books/{book_id}")
async def delete_book_by_id(book_id : int = Path(gt=0)):
# it will make sure that index will start from 1.
for i in range(len(BOOKS)):
if BOOKS[i].id == book_id:
BOOKS.pop(i)
return {"message": "Book is deleted successfully."}
return {"message": "Book not found."}
Result:
Query Parameter Data Validation
It is similar to Path parameter data validation just need to replace Path keyword with Query keyword.
Code:
from fastapi import FastAPI, Path, Query
from books3 import Book
from datavalidatorhelper import BookRequest
app = FastAPI()
BOOKS = [
Book(id=1, title="DSA with Java",
author="Puneet Kumar Singh",
description="A comprehensive guide to Data Structures
and Algorithms using Java.",
rating=5,
published_date=2014),
Book(id=2, title="Python Programming",
author="Puneet Kumar Singh",
description="An introduction to Python programming for beginners.",
rating=4,
published_date=2015),
Book(id=3, title="Machine Learning Basics",
author="Puneet Kumar Singh",
description="Understanding the fundamentals of Machine Learning.",
rating=4,
published_date=2014),
Book(id=4, title="Web Development with FastAPI",
author="Ramesh Jha",
description="Building web applications using FastAPI.",
rating=3,
published_date=2016),
Book(id=5, title="Data Science with Python",
author="Kalidas",
description="A guide to Data Science using Python.",
rating=2,
published_date=2018),
Book(id=6, title="Advanced Java Programming",
author="Kalpana Gupta",
description="Deep dive into advanced concepts of Java programming.",
rating=1,
published_date=2014)
]
# API endpoint to get all books detail unfiltered.
@app.get("/books") # static path to get all books
async def read_all_books():
"""
This function returns the list of all books.
"""
return BOOKS
""" post request without any input validation
@app.post("/create_book")
async def create_book(book_request=Body()):
BOOKS.append(book_request)
"""
#API endpoint to fetch book by using published_date as parameter
@app.get("/books/published/")
async def read_book_by_published_date(published_date: int = Query(gt=1999, lt=2031)):
# contraints check at the time of input
books_to_return = []
for book in BOOKS:
if book.published_date == published_date:
books_to_return.append(book)
return books_to_return
# API endpoint to fetch book by using id
@app.get("/books/{book_id}")
async def read_book_by_id(book_id: int = Path(gt=0)):
# it will make sure that index will start from 1.
for book in BOOKS:
if book.id == book_id:
return book
return {"message": "book not found"}
# API endpoint to fetch books by rating
@app.get("/books/")
async def read_book_by_rating(rating: int = Query(gt=0, lt=6)):
# contraints check at the time of input
book_to_return = []
for book in BOOKS:
if book.rating == rating:
book_to_return.append(book)
return book_to_return
#post request with input validator
@app.post("/creat_book")
async def create_book(book_request: BookRequest):
new_book = Book(**book_request.model_dump())
# new_book = Book(**book_request.dict()) -> if you are using
pydantic version 1 else the model_dump() for pydentic version 2.
# BOOKS.append(new_book) -> it will store the valid input
dump but you can have duplicate ids.
BOOKS.append(find_book_id(new_book))
# It will add latest id sequence number to id automatically
and will overwrite the provided input so you don't need
to bother about the id value sequence.
def find_book_id(book: Book):
if len(BOOKS) > 0:
book.id = BOOKS[-1].id + 1
else:
book.id = 1
return book
# API endpoint to update existing book entry
@app.put("/books/update_book")
async def update_book_by_id(book_request: BookRequest):
new_book=Book(**book_request.model_dump())
for i in range(len(BOOKS)):
if BOOKS[i].id == new_book.id:
BOOKS[i]=new_book
return {"message": "Book is updated", "detail": new_book}
return {"message": "Book id not found. Hence can't add
new book to library. Do use create book API endpoint to add new book into library."}
# API endpoint to delete book entry from library
@app.delete("/books/{book_id}")
async def delete_book_by_id(book_id : int = Path(gt=0)):
# it will make sure that index will start from 1.
for i in range(len(BOOKS)):
if BOOKS[i].id == book_id:
BOOKS.pop(i)
return {"message": "Book is deleted successfully."}
return {"message": "Book not found."}
Result:
Comments
Post a Comment