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
Fieldfor fine-grained control over field validation. You can specify things like minimum and maximum values, regex patterns, and more. -
Take advantage of Pydantic’s
Configclass 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
Optionalfields for attributes that aren’t required. -
Leverage FastAPI’s background tasks for operations that don’t need to block the response.
-
Use FastAPI’s
APIRouterto 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
Fieldvalidators in our Pydantic models. -
We’ve added a simple authentication system using OAuth2PasswordBearer and a
get_current_userdependency. -
We’re using
asyncfunctions 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
