FastAPI is a modern, high-performance web framework for building APIs with Python. When combined with Pydantic, it becomes a powerhouse for data validation and serialization. Let’s dive into how you can leverage these tools to create robust and efficient APIs.
First things first, you’ll need to install FastAPI and Pydantic. Open up your terminal and run:
pip install fastapi pydantic
Now that we’ve got the basics out of the way, let’s start building our API. We’ll create a simple book management system to demonstrate the concepts.
Here’s our main FastAPI application:
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from typing import List, Optional
import uuid
app = FastAPI()
class Book(BaseModel):
id: str = Field(default_factory=lambda: str(uuid.uuid4()))
title: str
author: str
description: Optional[str] = None
price: float = Field(gt=0)
class Config:
schema_extra = {
"example": {
"title": "To Kill a Mockingbird",
"author": "Harper Lee",
"description": "A classic of modern American literature",
"price": 9.99
}
}
books = []
@app.post("/books", response_model=Book)
def create_book(book: Book):
books.append(book)
return book
@app.get("/books", response_model=List[Book])
def get_books():
return books
@app.get("/books/{book_id}", response_model=Book)
def get_book(book_id: str):
for book in books:
if book.id == book_id:
return book
raise HTTPException(status_code=404, detail="Book not found")
@app.put("/books/{book_id}", response_model=Book)
def update_book(book_id: str, updated_book: Book):
for i, book in enumerate(books):
if book.id == book_id:
books[i] = updated_book
return updated_book
raise HTTPException(status_code=404, detail="Book not found")
@app.delete("/books/{book_id}")
def delete_book(book_id: str):
for i, book in enumerate(books):
if book.id == book_id:
del books[i]
return {"message": "Book deleted successfully"}
raise HTTPException(status_code=404, detail="Book not found")
This code sets up a basic CRUD (Create, Read, Update, Delete) API for managing books. Let’s break it down and see how FastAPI and Pydantic work together to make our lives easier.
The Book
class is a Pydantic model that defines the structure of our book data. Pydantic handles data validation, serialization, and even generates JSON Schema documentation for us. Notice how we use Field
to add constraints and metadata to our fields.
The id
field uses a default factory to generate a unique UUID for each book. The price
field has a constraint that it must be greater than zero. The Config
class inside Book
provides an example that FastAPI will use in the automatic API documentation.
Our API endpoints use type hints and Pydantic models to specify request and response schemas. FastAPI uses these to validate incoming data and serialize outgoing data automatically.
For example, in the create_book
function, FastAPI will automatically parse the incoming JSON, validate it against the Book
model, and give us a Book
instance to work with. If the incoming data doesn’t match the Book
schema, FastAPI will return a detailed error message to the client.
The response_model
parameter in the route decorators tells FastAPI what the response should look like. This adds an extra layer of validation to ensure we’re sending back the correct data.
Now, let’s add some more advanced features to our API. We’ll implement pagination for the book list and add some query parameters for filtering:
from fastapi import FastAPI, HTTPException, Query
from pydantic import BaseModel, Field
from typing import List, Optional
import uuid
app = FastAPI()
# ... (previous Book model and other endpoints) ...
@app.get("/books", response_model=List[Book])
def get_books(
skip: int = Query(0, ge=0),
limit: int = Query(10, ge=1, le=100),
author: Optional[str] = None,
min_price: Optional[float] = None,
max_price: Optional[float] = None
):
filtered_books = books
if author:
filtered_books = [book for book in filtered_books if book.author.lower() == author.lower()]
if min_price is not None:
filtered_books = [book for book in filtered_books if book.price >= min_price]
if max_price is not None:
filtered_books = [book for book in filtered_books if book.price <= max_price]
return filtered_books[skip : skip + limit]
This updated get_books
endpoint now supports pagination with skip
and limit
parameters, as well as filtering by author and price range. FastAPI will automatically validate these query parameters based on the type hints and Query
validators we’ve provided.
Let’s add one more feature: a search endpoint that uses full-text search. For this example, we’ll use a simple string matching approach, but in a real-world scenario, you might want to use a more sophisticated search engine.
from fastapi import FastAPI, HTTPException, Query
from pydantic import BaseModel, Field
from typing import List, Optional
import uuid
app = FastAPI()
# ... (previous code) ...
@app.get("/books/search", response_model=List[Book])
def search_books(
query: str = Query(..., min_length=3),
skip: int = Query(0, ge=0),
limit: int = Query(10, ge=1, le=100)
):
results = []
for book in books:
if (query.lower() in book.title.lower() or
query.lower() in book.author.lower() or
(book.description and query.lower() in book.description.lower())):
results.append(book)
return results[skip : skip + limit]
This search endpoint allows users to search for books by title, author, or description. The query
parameter is required and must be at least 3 characters long, as specified by the Query
validator.
Now, let’s talk about some best practices when using FastAPI and Pydantic:
-
Use Pydantic’s
Field
for fine-grained control over field validation. You can specify things like minimum and maximum values, regex patterns, and more. -
Take advantage of Pydantic’s
Config
class to customize model behavior. You can use it to provide examples, configure case sensitivity, and more. -
Use type hints consistently. They not only provide better IDE support and catch errors early, but FastAPI also uses them for request validation and documentation generation.
-
Use FastAPI’s dependency injection system for things like authentication, database connections, and other shared resources.
-
Implement proper error handling using FastAPI’s
HTTPException
. -
Use Pydantic’s
Optional
fields for attributes that aren’t required. -
Leverage FastAPI’s background tasks for operations that don’t need to block the response.
-
Use FastAPI’s
APIRouter
to organize your endpoints into logical groups.
Let’s implement some of these best practices in our code:
from fastapi import FastAPI, HTTPException, Query, Depends, BackgroundTasks
from fastapi.security import OAuth2PasswordBearer
from pydantic import BaseModel, Field, EmailStr
from typing import List, Optional
import uuid
app = FastAPI()
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
class User(BaseModel):
id: str = Field(default_factory=lambda: str(uuid.uuid4()))
username: str = Field(..., min_length=3, max_length=50)
email: EmailStr
full_name: Optional[str] = None
class Book(BaseModel):
id: str = Field(default_factory=lambda: str(uuid.uuid4()))
title: str = Field(..., min_length=1, max_length=100)
author: str = Field(..., min_length=1, max_length=100)
description: Optional[str] = Field(None, max_length=1000)
price: float = Field(..., gt=0, le=1000)
owner_id: str
class Config:
schema_extra = {
"example": {
"title": "To Kill a Mockingbird",
"author": "Harper Lee",
"description": "A classic of modern American literature",
"price": 9.99,
"owner_id": "12345"
}
}
books = []
users = []
def get_current_user(token: str = Depends(oauth2_scheme)):
# In a real application, you would decode and verify the token here
# For this example, we'll just return a dummy user
return User(id="12345", username="johndoe", email="[email protected]")
@app.post("/books", response_model=Book)
async def create_book(book: Book, current_user: User = Depends(get_current_user), background_tasks: BackgroundTasks = BackgroundTasks()):
book.owner_id = current_user.id
books.append(book)
background_tasks.add_task(log_book_creation, book)
return book
@app.get("/books", response_model=List[Book])
async def get_books(
skip: int = Query(0, ge=0),
limit: int = Query(10, ge=1, le=100),
current_user: User = Depends(get_current_user)
):
return books[skip : skip + limit]
@app.get("/books/{book_id}", response_model=Book)
async def get_book(book_id: str, current_user: User = Depends(get_current_user)):
for book in books:
if book.id == book_id:
return book
raise HTTPException(status_code=404, detail="Book not found")
@app.put("/books/{book_id}", response_model=Book)
async def update_book(book_id: str, updated_book: Book, current_user: User = Depends(get_current_user)):
for i, book in enumerate(books):
if book.id == book_id:
if book.owner_id != current_user.id:
raise HTTPException(status_code=403, detail="Not authorized to update this book")
books[i] = updated_book
return updated_book
raise HTTPException(status_code=404, detail="Book not found")
@app.delete("/books/{book_id}")
async def delete_book(book_id: str, current_user: User = Depends(get_current_user)):
for i, book in enumerate(books):
if book.id == book_id:
if book.owner_id != current_user.id:
raise HTTPException(status_code=403, detail="Not authorized to delete this book")
del books[i]
return {"message": "Book deleted successfully"}
raise HTTPException(status_code=404, detail="Book not found")
async def log_book_creation(book: Book):
# In a real application, this might write to a database or send a notification
print(f"Book created: {book.title}")
In this updated version, we’ve implemented several best practices:
-
We’re using more detailed
Field
validators in our Pydantic models. -
We’ve added a simple authentication system using OAuth2PasswordBearer and a
get_current_user
dependency. -
We’re using
async
functions for all our endpoints, which can improve performance for I/O-bound operations. -
We’ve added owner-based authorization for update and delete operations.
-
We’re using a background task to log book creation, demonstrating how to perform non-blocking operations.
This example showcases how FastAPI and Pydantic work together to create a robust, type-safe API with minimal boilerplate code. The built-in data validation