Building a REST API is one of the most common use cases for FastAPI. Its combination of automatic validation, serialization, and interactive documentation makes it the ideal framework for creating robust, production-ready APIs. In this tutorial, we will cover everything you need to know to build a complete REST API with FastAPI — from fundamental REST principles to advanced patterns like pagination, filtering, versioning, background tasks, and middleware.
pip install fastapi uvicorn sqlalchemy pydantic[email] python-multipart
REST (Representational State Transfer) is an architectural style for designing networked applications. It was introduced by Roy Fielding in his 2000 doctoral dissertation and has become the dominant approach for building web APIs. Understanding REST principles is essential before writing any API code.
A truly RESTful API adheres to six architectural constraints:
| Constraint | Description | Benefit |
|---|---|---|
| Client-Server | Separation of concerns between UI and data storage | Independent evolution of client and server |
| Stateless | Each request contains all information needed to process it | Scalability — any server can handle any request |
| Cacheable | Responses must define themselves as cacheable or not | Reduced latency and server load |
| Uniform Interface | Standard methods (GET, POST, PUT, DELETE) on resources | Simplicity and decoupling |
| Layered System | Client cannot tell if connected directly to server | Load balancing, caching proxies, security |
| Code on Demand (optional) | Server can extend client functionality with code | Reduced pre-implemented features needed |
In REST, everything is a resource. A resource is any concept that can be addressed — a user, a book, an order, a collection of items. Each resource is identified by a URI (Uniform Resource Identifier).
Good REST URI design follows these conventions:
| Pattern | Example | Description |
|---|---|---|
| Collection | /api/v1/books |
All books (plural noun) |
| Single resource | /api/v1/books/42 |
Book with ID 42 |
| Sub-resource | /api/v1/books/42/reviews |
Reviews belonging to book 42 |
| Filtered collection | /api/v1/books?genre=fiction |
Books filtered by genre |
| Nested sub-resource | /api/v1/users/5/orders/10 |
Order 10 of user 5 |
/api/getBooks — use /api/books with GET method instead/api/book — use /api/books (collections are plural)/api/v1/users/5/orders/10/items/3/reviews — flatten it/api/books/delete/42 — use DELETE /api/books/42 insteadREST maps CRUD operations to HTTP methods. Each method has specific semantics regarding safety and idempotency:
| Method | CRUD | Safe | Idempotent | Request Body | Typical Status Codes |
|---|---|---|---|---|---|
| GET | Read | Yes | Yes | No | 200, 404 |
| POST | Create | No | No | Yes | 201, 400, 409 |
| PUT | Replace | No | Yes | Yes | 200, 204, 404 |
| PATCH | Partial Update | No | No | Yes | 200, 204, 404 |
| DELETE | Delete | No | Yes | Optional | 204, 404 |
| HEAD | Read (headers only) | Yes | Yes | No | 200, 404 |
| OPTIONS | Metadata | Yes | Yes | No | 200, 204 |
Safe means the method does not modify server state. Idempotent means calling it multiple times produces the same result as calling it once. Note that POST is neither safe nor idempotent — calling POST /books twice creates two books.
FastAPI maps naturally to REST principles. Here is a minimal example showing how REST concepts translate to FastAPI code:
from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel
from typing import Optional
app = FastAPI(title="REST Principles Demo")
# In-memory storage for demonstration
books_db: dict[int, dict] = {}
next_id: int = 1
class BookCreate(BaseModel):
title: str
author: str
isbn: str
class BookUpdate(BaseModel):
title: Optional[str] = None
author: Optional[str] = None
isbn: Optional[str] = None
# GET /books - Read collection (Safe, Idempotent)
@app.get("/books")
def list_books():
return list(books_db.values())
# GET /books/{id} - Read single resource (Safe, Idempotent)
@app.get("/books/{book_id}")
def get_book(book_id: int):
if book_id not in books_db:
raise HTTPException(status_code=404, detail="Book not found")
return books_db[book_id]
# POST /books - Create resource (Not safe, Not idempotent)
@app.post("/books", status_code=status.HTTP_201_CREATED)
def create_book(book: BookCreate):
global next_id
book_data = {"id": next_id, **book.model_dump()}
books_db[next_id] = book_data
next_id += 1
return book_data
# PUT /books/{id} - Replace resource (Not safe, Idempotent)
@app.put("/books/{book_id}")
def replace_book(book_id: int, book: BookCreate):
if book_id not in books_db:
raise HTTPException(status_code=404, detail="Book not found")
books_db[book_id] = {"id": book_id, **book.model_dump()}
return books_db[book_id]
# PATCH /books/{id} - Partial update (Not safe, Not idempotent)
@app.patch("/books/{book_id}")
def update_book(book_id: int, book: BookUpdate):
if book_id not in books_db:
raise HTTPException(status_code=404, detail="Book not found")
stored = books_db[book_id]
update_data = book.model_dump(exclude_unset=True)
stored.update(update_data)
return stored
# DELETE /books/{id} - Delete resource (Not safe, Idempotent)
@app.delete("/books/{book_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_book(book_id: int):
if book_id not in books_db:
raise HTTPException(status_code=404, detail="Book not found")
del books_db[book_id]
This example demonstrates the fundamental REST pattern: resources (books) are accessed via a uniform interface (HTTP methods) at consistent URIs. FastAPI handles JSON serialization, request validation, and documentation automatically.
Pydantic models are the backbone of request and response handling in FastAPI. They provide automatic validation, serialization, and documentation generation. A well-designed set of Pydantic models is the difference between a fragile API and a robust one.
A common best practice is to separate your models by purpose. You typically need different schemas for creating, updating, and reading a resource:
from pydantic import BaseModel, Field, EmailStr, ConfigDict
from typing import Optional
from datetime import datetime
from enum import Enum
class UserRole(str, Enum):
ADMIN = "admin"
EDITOR = "editor"
VIEWER = "viewer"
# Base schema — shared fields
class UserBase(BaseModel):
username: str = Field(
...,
min_length=3,
max_length=50,
pattern=r"^[a-zA-Z0-9_]+$",
description="Alphanumeric username, 3-50 characters"
)
email: EmailStr = Field(..., description="Valid email address")
full_name: str = Field(..., min_length=1, max_length=100)
role: UserRole = Field(default=UserRole.VIEWER)
# Create schema — fields needed for creation
class UserCreate(UserBase):
password: str = Field(
...,
min_length=8,
max_length=128,
description="Password must be at least 8 characters"
)
# Update schema — all fields optional
class UserUpdate(BaseModel):
username: Optional[str] = Field(
None, min_length=3, max_length=50, pattern=r"^[a-zA-Z0-9_]+$"
)
email: Optional[EmailStr] = None
full_name: Optional[str] = Field(None, min_length=1, max_length=100)
role: Optional[UserRole] = None
# Response schema — what the API returns (never includes password)
class UserResponse(UserBase):
id: int
is_active: bool = True
created_at: datetime
updated_at: Optional[datetime] = None
model_config = ConfigDict(from_attributes=True)
# List response with metadata
class UserListResponse(BaseModel):
users: list[UserResponse]
total: int
page: int
per_page: int
This pattern — Base, Create, Update, Response — keeps your API clean and secure. The password field only appears in UserCreate, never in responses.
Real-world APIs frequently deal with nested data. Pydantic handles this naturally:
from pydantic import BaseModel, Field
from typing import Optional
from datetime import datetime
class Address(BaseModel):
street: str
city: str
state: str = Field(..., min_length=2, max_length=2)
zip_code: str = Field(..., pattern=r"^\d{5}(-\d{4})?$")
country: str = "US"
class ContactInfo(BaseModel):
phone: Optional[str] = Field(
None, pattern=r"^\+?1?\d{10,15}$"
)
address: Optional[Address] = None
class AuthorCreate(BaseModel):
name: str = Field(..., min_length=1, max_length=200)
bio: Optional[str] = Field(None, max_length=2000)
contact: Optional[ContactInfo] = None
class BookCreate(BaseModel):
title: str = Field(..., min_length=1, max_length=300)
isbn: str = Field(..., pattern=r"^\d{13}$")
price: float = Field(..., gt=0, le=10000)
author: AuthorCreate
tags: list[str] = Field(default_factory=list, max_length=10)
metadata: dict[str, str] = Field(default_factory=dict)
class BookResponse(BaseModel):
id: int
title: str
isbn: str
price: float
author: AuthorCreate
tags: list[str]
created_at: datetime
model_config = ConfigDict(from_attributes=True)
FastAPI will validate the entire nested structure automatically. If address.zip_code does not match the regex, the client receives a clear 422 error with the exact path to the invalid field.
Pydantic v2 provides powerful validation through Field constraints and custom validators:
from pydantic import BaseModel, Field, field_validator, model_validator
from typing import Optional
from datetime import date
class EventCreate(BaseModel):
name: str = Field(..., min_length=3, max_length=200)
description: Optional[str] = Field(None, max_length=5000)
start_date: date
end_date: date
max_attendees: int = Field(default=100, ge=1, le=10000)
ticket_price: float = Field(default=0.0, ge=0)
tags: list[str] = Field(default_factory=list)
# Field-level validator
@field_validator("name")
@classmethod
def name_must_not_be_blank(cls, v: str) -> str:
if not v.strip():
raise ValueError("Name cannot be blank or whitespace only")
return v.strip()
@field_validator("tags")
@classmethod
def validate_tags(cls, v: list[str]) -> list[str]:
if len(v) > 10:
raise ValueError("Maximum 10 tags allowed")
# Normalize tags to lowercase
return [tag.lower().strip() for tag in v if tag.strip()]
# Model-level validator — access multiple fields
@model_validator(mode="after")
def validate_dates(self) -> "EventCreate":
if self.end_date < self.start_date:
raise ValueError("end_date must be after start_date")
if self.start_date < date.today():
raise ValueError("start_date cannot be in the past")
return self
@model_validator(mode="after")
def validate_free_event_limit(self) -> "EventCreate":
if self.ticket_price == 0 and self.max_attendees > 500:
raise ValueError(
"Free events are limited to 500 attendees"
)
return self
Pydantic v2 supports computed fields that are included in the serialized output but not required as input:
from pydantic import BaseModel, Field, computed_field
from datetime import datetime, date
class ProductResponse(BaseModel):
id: int
name: str
price: float
discount_percent: float = 0.0
created_at: datetime
@computed_field
@property
def discounted_price(self) -> float:
return round(self.price * (1 - self.discount_percent / 100), 2)
@computed_field
@property
def is_new(self) -> bool:
days_since_created = (datetime.now() - self.created_at).days
return days_since_created < 30
@computed_field
@property
def price_tier(self) -> str:
if self.discounted_price < 10:
return "budget"
elif self.discounted_price < 50:
return "mid-range"
else:
return "premium"
# Usage in an endpoint
@app.get("/products/{product_id}", response_model=ProductResponse)
def get_product(product_id: int):
product = get_product_from_db(product_id)
return product # computed fields are auto-calculated
Control how your models behave with model_config:
from pydantic import BaseModel, ConfigDict
class StrictBookCreate(BaseModel):
model_config = ConfigDict(
# Raise error on extra fields
extra="forbid",
# Strip whitespace from strings
str_strip_whitespace=True,
# Validate field defaults
validate_default=True,
# Allow population from ORM objects
from_attributes=True,
# Custom JSON schema example
json_schema_extra={
"examples": [
{
"title": "The Great Gatsby",
"author": "F. Scott Fitzgerald",
"isbn": "9780743273565",
"price": 14.99,
}
]
},
)
title: str
author: str
isbn: str
price: float
The extra="forbid" setting is particularly useful — it rejects requests that include unexpected fields, preventing clients from accidentally sending data you do not handle.
CRUD — Create, Read, Update, Delete — forms the foundation of any REST API. In this section, we will build a complete CRUD implementation with proper error handling, validation, and response formatting. We will use an in-memory store for simplicity, but the patterns apply equally to database-backed APIs.
# models.py
from pydantic import BaseModel, Field, ConfigDict
from typing import Optional
from datetime import datetime
from enum import Enum
class BookStatus(str, Enum):
AVAILABLE = "available"
CHECKED_OUT = "checked_out"
RESERVED = "reserved"
ARCHIVED = "archived"
class BookBase(BaseModel):
title: str = Field(..., min_length=1, max_length=300)
author: str = Field(..., min_length=1, max_length=200)
isbn: str = Field(..., pattern=r"^\d{10}(\d{3})?$")
genre: Optional[str] = Field(None, max_length=50)
published_year: Optional[int] = Field(None, ge=1000, le=2030)
price: float = Field(..., gt=0, le=10000)
description: Optional[str] = Field(None, max_length=5000)
class BookCreate(BookBase):
"""Schema for creating a new book."""
pass
class BookUpdate(BaseModel):
"""Schema for partial updates — all fields optional."""
title: Optional[str] = Field(None, min_length=1, max_length=300)
author: Optional[str] = Field(None, min_length=1, max_length=200)
isbn: Optional[str] = Field(None, pattern=r"^\d{10}(\d{3})?$")
genre: Optional[str] = Field(None, max_length=50)
published_year: Optional[int] = Field(None, ge=1000, le=2030)
price: Optional[float] = Field(None, gt=0, le=10000)
description: Optional[str] = Field(None, max_length=5000)
status: Optional[BookStatus] = None
class BookResponse(BookBase):
"""Schema for book responses."""
id: int
status: BookStatus = BookStatus.AVAILABLE
created_at: datetime
updated_at: Optional[datetime] = None
model_config = ConfigDict(from_attributes=True)
# store.py
from datetime import datetime
from typing import Optional
class BookStore:
"""Simple in-memory data store for books."""
def __init__(self):
self._books: dict[int, dict] = {}
self._next_id: int = 1
def create(self, data: dict) -> dict:
book_id = self._next_id
self._next_id += 1
book = {
"id": book_id,
**data,
"status": "available",
"created_at": datetime.utcnow(),
"updated_at": None,
}
self._books[book_id] = book
return book
def get(self, book_id: int) -> Optional[dict]:
return self._books.get(book_id)
def get_all(self) -> list[dict]:
return list(self._books.values())
def update(self, book_id: int, data: dict) -> Optional[dict]:
if book_id not in self._books:
return None
book = self._books[book_id]
book.update(data)
book["updated_at"] = datetime.utcnow()
return book
def delete(self, book_id: int) -> bool:
if book_id not in self._books:
return False
del self._books[book_id]
return True
def exists(self, book_id: int) -> bool:
return book_id in self._books
def find_by_isbn(self, isbn: str) -> Optional[dict]:
for book in self._books.values():
if book["isbn"] == isbn:
return book
return None
def count(self) -> int:
return len(self._books)
# Global instance
book_store = BookStore()
from fastapi import FastAPI, HTTPException, status, Response
from models import BookCreate, BookUpdate, BookResponse
from store import book_store
app = FastAPI(title="Book Management API")
@app.post(
"/api/v1/books",
response_model=BookResponse,
status_code=status.HTTP_201_CREATED,
summary="Create a new book",
description="Add a new book to the catalog. ISBN must be unique.",
)
def create_book(book: BookCreate):
# Check for duplicate ISBN
existing = book_store.find_by_isbn(book.isbn)
if existing:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail=f"Book with ISBN {book.isbn} already exists (ID: {existing['id']})",
)
book_data = book.model_dump()
created_book = book_store.create(book_data)
return created_book
Key points about the Create endpoint:
response_model=BookResponse ensures the response matches our schemafrom typing import Optional
@app.get(
"/api/v1/books",
response_model=list[BookResponse],
summary="List all books",
)
def list_books(
genre: Optional[str] = None,
author: Optional[str] = None,
status_filter: Optional[str] = None,
):
"""Retrieve all books, optionally filtered by genre, author, or status."""
books = book_store.get_all()
if genre:
books = [b for b in books if b.get("genre", "").lower() == genre.lower()]
if author:
books = [b for b in books if author.lower() in b["author"].lower()]
if status_filter:
books = [b for b in books if b["status"] == status_filter]
return books
@app.get(
"/api/v1/books/{book_id}",
response_model=BookResponse,
summary="Get a specific book",
)
def get_book(book_id: int):
"""Retrieve a single book by its ID."""
book = book_store.get(book_id)
if not book:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Book with ID {book_id} not found",
)
return book
REST distinguishes between PUT (full replacement) and PATCH (partial update). Both are important:
# PUT — Full replacement (all required fields must be provided)
@app.put(
"/api/v1/books/{book_id}",
response_model=BookResponse,
summary="Replace a book",
)
def replace_book(book_id: int, book: BookCreate):
"""Replace all fields of an existing book."""
if not book_store.exists(book_id):
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Book with ID {book_id} not found",
)
# Check ISBN conflict (but allow same book to keep its ISBN)
existing = book_store.find_by_isbn(book.isbn)
if existing and existing["id"] != book_id:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail=f"ISBN {book.isbn} already belongs to book ID {existing['id']}",
)
updated = book_store.update(book_id, book.model_dump())
return updated
# PATCH — Partial update (only provided fields are updated)
@app.patch(
"/api/v1/books/{book_id}",
response_model=BookResponse,
summary="Update a book partially",
)
def update_book(book_id: int, book: BookUpdate):
"""Update specific fields of an existing book."""
if not book_store.exists(book_id):
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Book with ID {book_id} not found",
)
# Only include fields that were explicitly set
update_data = book.model_dump(exclude_unset=True)
if not update_data:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="No fields provided for update",
)
# Check ISBN conflict if ISBN is being updated
if "isbn" in update_data:
existing = book_store.find_by_isbn(update_data["isbn"])
if existing and existing["id"] != book_id:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail=f"ISBN {update_data['isbn']} already belongs to book ID {existing['id']}",
)
updated = book_store.update(book_id, update_data)
return updated
None vs fields not provided at all are handled differently — model_dump(exclude_unset=True) only returns fields the client actually sent.@app.delete(
"/api/v1/books/{book_id}",
status_code=status.HTTP_204_NO_CONTENT,
summary="Delete a book",
)
def delete_book(book_id: int):
"""Remove a book from the catalog."""
if not book_store.delete(book_id):
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Book with ID {book_id} not found",
)
# 204 No Content — return nothing
return Response(status_code=status.HTTP_204_NO_CONTENT)
Production APIs often need bulk create and delete operations:
from pydantic import BaseModel
class BulkCreateResponse(BaseModel):
created: list[BookResponse]
errors: list[dict]
@app.post(
"/api/v1/books/bulk",
response_model=BulkCreateResponse,
status_code=status.HTTP_201_CREATED,
summary="Create multiple books",
)
def bulk_create_books(books: list[BookCreate]):
"""Create multiple books in a single request. Returns created books and any errors."""
if len(books) > 100:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Maximum 100 books per bulk request",
)
created = []
errors = []
for i, book in enumerate(books):
existing = book_store.find_by_isbn(book.isbn)
if existing:
errors.append({
"index": i,
"isbn": book.isbn,
"error": f"Duplicate ISBN (existing ID: {existing['id']})",
})
continue
book_data = book.model_dump()
created_book = book_store.create(book_data)
created.append(created_book)
return {"created": created, "errors": errors}
class BulkDeleteRequest(BaseModel):
ids: list[int] = Field(..., min_length=1, max_length=100)
class BulkDeleteResponse(BaseModel):
deleted: list[int]
not_found: list[int]
@app.post(
"/api/v1/books/bulk-delete",
response_model=BulkDeleteResponse,
summary="Delete multiple books",
)
def bulk_delete_books(request: BulkDeleteRequest):
"""Delete multiple books by ID. Returns which IDs were deleted and which were not found."""
deleted = []
not_found = []
for book_id in request.ids:
if book_store.delete(book_id):
deleted.append(book_id)
else:
not_found.append(book_id)
return {"deleted": deleted, "not_found": not_found}
Note that we use POST /books/bulk-delete instead of DELETE /books because DELETE with a request body is not universally supported by all HTTP clients and proxies.
FastAPI’s response_model parameter gives you fine-grained control over what data is returned to clients. It filters out sensitive fields, adds computed properties, and generates accurate OpenAPI documentation.
from fastapi import FastAPI
from pydantic import BaseModel, EmailStr, ConfigDict
from typing import Optional
from datetime import datetime
app = FastAPI()
class UserInternal(BaseModel):
"""Internal user model — contains sensitive data."""
id: int
username: str
email: str
hashed_password: str
api_key: str
is_active: bool
login_attempts: int
created_at: datetime
class UserPublic(BaseModel):
"""Public user model — safe to return to clients."""
id: int
username: str
email: str
is_active: bool
created_at: datetime
model_config = ConfigDict(from_attributes=True)
# The response_model filters out sensitive fields automatically
@app.get("/users/{user_id}", response_model=UserPublic)
def get_user(user_id: int):
# Even though our internal data includes hashed_password and api_key,
# the response_model ensures they are never sent to the client
user = get_user_from_db(user_id) # Returns UserInternal
return user
Even if the function returns an object with extra fields (like hashed_password), the response_model strips them from the response. This is a critical security feature.
For quick filtering without creating separate models, use response_model_exclude and response_model_include:
class Book(BaseModel):
id: int
title: str
author: str
isbn: str
price: float
description: str
internal_notes: str
created_at: datetime
# Exclude specific fields
@app.get(
"/books/{book_id}",
response_model=Book,
response_model_exclude={"internal_notes"},
)
def get_book(book_id: int):
return books_db[book_id]
# Include only specific fields (useful for summary views)
@app.get(
"/books/{book_id}/summary",
response_model=Book,
response_model_include={"id", "title", "author", "price"},
)
def get_book_summary(book_id: int):
return books_db[book_id]
# Exclude None values from response
@app.get(
"/books/{book_id}/compact",
response_model=Book,
response_model_exclude_none=True,
)
def get_book_compact(book_id: int):
"""Only include fields that have non-None values."""
return books_db[book_id]
# Exclude default values
@app.get(
"/books/{book_id}/minimal",
response_model=Book,
response_model_exclude_defaults=True,
)
def get_book_minimal(book_id: int):
"""Only include fields that differ from their defaults."""
return books_db[book_id]
Different endpoints often need different views of the same resource:
from pydantic import BaseModel, ConfigDict
from typing import Optional
from datetime import datetime
class BookSummary(BaseModel):
"""Minimal view for list endpoints."""
id: int
title: str
author: str
price: float
model_config = ConfigDict(from_attributes=True)
class BookDetail(BaseModel):
"""Full view for detail endpoints."""
id: int
title: str
author: str
isbn: str
price: float
genre: Optional[str]
published_year: Optional[int]
description: Optional[str]
page_count: Optional[int]
created_at: datetime
updated_at: Optional[datetime]
model_config = ConfigDict(from_attributes=True)
class BookAdmin(BookDetail):
"""Admin view with internal fields."""
internal_notes: Optional[str]
cost_price: float
profit_margin: float
supplier: Optional[str]
# List endpoint returns summaries
@app.get("/api/v1/books", response_model=list[BookSummary])
def list_books():
return book_store.get_all()
# Detail endpoint returns full data
@app.get("/api/v1/books/{book_id}", response_model=BookDetail)
def get_book(book_id: int):
return book_store.get(book_id)
# Admin endpoint returns everything
@app.get("/api/v1/admin/books/{book_id}", response_model=BookAdmin)
def get_book_admin(book_id: int):
return book_store.get(book_id)
Most production APIs wrap their responses in a standard envelope for consistency:
from pydantic import BaseModel
from typing import TypeVar, Generic, Optional
from datetime import datetime
T = TypeVar("T")
class APIResponse(BaseModel, Generic[T]):
"""Standard API response envelope."""
success: bool = True
data: T
message: Optional[str] = None
timestamp: datetime = Field(default_factory=datetime.utcnow)
class PaginatedResponse(BaseModel, Generic[T]):
"""Paginated response with metadata."""
items: list[T]
total: int
page: int
per_page: int
total_pages: int
class ErrorResponse(BaseModel):
"""Standard error response."""
success: bool = False
error: str
detail: Optional[str] = None
timestamp: datetime = Field(default_factory=datetime.utcnow)
# Usage
@app.get(
"/api/v1/books/{book_id}",
response_model=APIResponse[BookDetail],
)
def get_book(book_id: int):
book = book_store.get(book_id)
if not book:
raise HTTPException(status_code=404, detail="Book not found")
return APIResponse(data=book, message="Book retrieved successfully")
@app.get(
"/api/v1/books",
response_model=APIResponse[PaginatedResponse[BookSummary]],
)
def list_books(page: int = 1, per_page: int = 20):
all_books = book_store.get_all()
total = len(all_books)
start = (page - 1) * per_page
end = start + per_page
items = all_books[start:end]
total_pages = (total + per_page - 1) // per_page
paginated = PaginatedResponse(
items=items,
total=total,
page=page,
per_page=per_page,
total_pages=total_pages,
)
return APIResponse(data=paginated)
Using generic wrapper models gives your API a consistent feel. Clients always know the shape of the response — success, data, message — regardless of the endpoint.
Proper HTTP status codes and error handling are what separate amateur APIs from professional ones. Clients depend on status codes for control flow, and clear error messages reduce debugging time dramatically.
| Code | Name | When to Use |
|---|---|---|
| 200 | OK | Successful GET, PUT, PATCH |
| 201 | Created | Successful POST that created a resource |
| 204 | No Content | Successful DELETE |
| 400 | Bad Request | Invalid request syntax or parameters |
| 401 | Unauthorized | Missing or invalid authentication |
| 403 | Forbidden | Authenticated but not authorized |
| 404 | Not Found | Resource does not exist |
| 409 | Conflict | Duplicate resource (e.g., duplicate email) |
| 422 | Unprocessable Entity | Validation error (FastAPI default) |
| 429 | Too Many Requests | Rate limit exceeded |
| 500 | Internal Server Error | Unexpected server error |
FastAPI provides HTTPException for raising errors with appropriate status codes:
from fastapi import FastAPI, HTTPException, status
app = FastAPI()
@app.get("/books/{book_id}")
def get_book(book_id: int):
book = book_store.get(book_id)
if not book:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Book not found",
headers={"X-Error": "Book lookup failed"},
)
return book
@app.post("/books")
def create_book(book: BookCreate):
# Business logic validation
if book.price > 1000 and not book.isbn.startswith("978"):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail={
"message": "Premium books must have a valid ISBN-13",
"field": "isbn",
"constraint": "Must start with 978 for books over $1000",
},
)
return book_store.create(book.model_dump())
The detail parameter can be a string or a dictionary — use dictionaries for structured error responses.
For larger applications, define custom exception classes and register handlers:
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from datetime import datetime
# Custom exception classes
class AppException(Exception):
"""Base exception for the application."""
def __init__(self, message: str, status_code: int = 500):
self.message = message
self.status_code = status_code
class NotFoundException(AppException):
def __init__(self, resource: str, resource_id):
super().__init__(
message=f"{resource} with ID {resource_id} not found",
status_code=404,
)
self.resource = resource
self.resource_id = resource_id
class DuplicateException(AppException):
def __init__(self, resource: str, field: str, value: str):
super().__init__(
message=f"{resource} with {field}='{value}' already exists",
status_code=409,
)
class ValidationException(AppException):
def __init__(self, errors: list[dict]):
super().__init__(message="Validation failed", status_code=422)
self.errors = errors
class RateLimitException(AppException):
def __init__(self, retry_after: int = 60):
super().__init__(
message="Rate limit exceeded",
status_code=429,
)
self.retry_after = retry_after
app = FastAPI()
# Register exception handlers
@app.exception_handler(AppException)
async def app_exception_handler(request: Request, exc: AppException):
return JSONResponse(
status_code=exc.status_code,
content={
"success": False,
"error": type(exc).__name__,
"message": exc.message,
"path": str(request.url),
"timestamp": datetime.utcnow().isoformat(),
},
)
@app.exception_handler(NotFoundException)
async def not_found_handler(request: Request, exc: NotFoundException):
return JSONResponse(
status_code=404,
content={
"success": False,
"error": "NotFound",
"message": exc.message,
"resource": exc.resource,
"resource_id": str(exc.resource_id),
"path": str(request.url),
"timestamp": datetime.utcnow().isoformat(),
},
)
@app.exception_handler(RateLimitException)
async def rate_limit_handler(request: Request, exc: RateLimitException):
return JSONResponse(
status_code=429,
content={
"success": False,
"error": "RateLimitExceeded",
"message": exc.message,
"retry_after": exc.retry_after,
},
headers={"Retry-After": str(exc.retry_after)},
)
# Usage in endpoints
@app.get("/books/{book_id}")
def get_book(book_id: int):
book = book_store.get(book_id)
if not book:
raise NotFoundException("Book", book_id)
return book
@app.post("/books", status_code=201)
def create_book(book: BookCreate):
existing = book_store.find_by_isbn(book.isbn)
if existing:
raise DuplicateException("Book", "isbn", book.isbn)
return book_store.create(book.model_dump())
FastAPI returns 422 errors for validation failures, but the default format may not match your API style. You can override it:
from fastapi import FastAPI, Request
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from starlette.exceptions import HTTPException as StarletteHTTPException
app = FastAPI()
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(
request: Request, exc: RequestValidationError
):
"""Transform Pydantic validation errors into a cleaner format."""
errors = []
for error in exc.errors():
field_path = " -> ".join(str(loc) for loc in error["loc"])
errors.append({
"field": field_path,
"message": error["msg"],
"type": error["type"],
})
return JSONResponse(
status_code=422,
content={
"success": False,
"error": "ValidationError",
"message": "Request validation failed",
"errors": errors,
"body": exc.body,
},
)
@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(
request: Request, exc: StarletteHTTPException
):
"""Standardize all HTTP error responses."""
return JSONResponse(
status_code=exc.status_code,
content={
"success": False,
"error": "HTTPError",
"message": str(exc.detail),
"status_code": exc.status_code,
},
)
@app.exception_handler(Exception)
async def general_exception_handler(request: Request, exc: Exception):
"""Catch-all handler for unexpected errors."""
# Log the error for debugging (never expose details to clients)
import logging
logging.exception(f"Unhandled error at {request.url}")
return JSONResponse(
status_code=500,
content={
"success": False,
"error": "InternalServerError",
"message": "An unexpected error occurred",
},
)
Document your error responses so API consumers know what to expect:
from pydantic import BaseModel
class ErrorDetail(BaseModel):
success: bool = False
error: str
message: str
@app.get(
"/books/{book_id}",
response_model=BookResponse,
responses={
200: {"description": "Book found successfully"},
404: {
"model": ErrorDetail,
"description": "Book not found",
"content": {
"application/json": {
"example": {
"success": False,
"error": "NotFound",
"message": "Book with ID 42 not found",
}
}
},
},
422: {
"model": ErrorDetail,
"description": "Validation error",
},
},
)
def get_book(book_id: int):
book = book_store.get(book_id)
if not book:
raise NotFoundException("Book", book_id)
return book
Any API that returns collections must implement pagination. Without it, a list endpoint could return millions of records, overwhelming both the server and client. FastAPI makes it straightforward to implement multiple pagination strategies.
The simplest and most common pagination style uses offset (or skip) and limit parameters:
from fastapi import FastAPI, Query
from pydantic import BaseModel
from typing import TypeVar, Generic
from math import ceil
app = FastAPI()
T = TypeVar("T")
class PaginatedResponse(BaseModel, Generic[T]):
items: list[T]
total: int
page: int
per_page: int
total_pages: int
has_next: bool
has_prev: bool
def paginate(
items: list,
page: int,
per_page: int,
) -> dict:
"""Generic pagination helper."""
total = len(items)
total_pages = ceil(total / per_page) if per_page > 0 else 0
start = (page - 1) * per_page
end = start + per_page
return {
"items": items[start:end],
"total": total,
"page": page,
"per_page": per_page,
"total_pages": total_pages,
"has_next": page < total_pages,
"has_prev": page > 1,
}
@app.get("/api/v1/books", response_model=PaginatedResponse[BookSummary])
def list_books(
page: int = Query(1, ge=1, description="Page number"),
per_page: int = Query(20, ge=1, le=100, description="Items per page"),
):
"""List books with offset/limit pagination."""
all_books = book_store.get_all()
return paginate(all_books, page, per_page)
To avoid repeating pagination parameters on every endpoint, create a reusable dependency:
from fastapi import Depends, Query
from dataclasses import dataclass
@dataclass
class PaginationParams:
"""Reusable pagination parameters."""
page: int = Query(1, ge=1, description="Page number (1-based)")
per_page: int = Query(20, ge=1, le=100, description="Items per page")
@property
def offset(self) -> int:
return (self.page - 1) * self.per_page
@property
def limit(self) -> int:
return self.per_page
@app.get("/api/v1/books")
def list_books(pagination: PaginationParams = Depends()):
all_books = book_store.get_all()
total = len(all_books)
items = all_books[pagination.offset : pagination.offset + pagination.limit]
return {
"items": items,
"total": total,
"page": pagination.page,
"per_page": pagination.per_page,
"total_pages": ceil(total / pagination.per_page),
}
@app.get("/api/v1/authors")
def list_authors(pagination: PaginationParams = Depends()):
# Same pagination logic, different resource
all_authors = author_store.get_all()
total = len(all_authors)
items = all_authors[pagination.offset : pagination.offset + pagination.limit]
return {
"items": items,
"total": total,
"page": pagination.page,
"per_page": pagination.per_page,
}
Cursor-based pagination is more efficient for large datasets and avoids the “shifting window” problem of offset pagination (where new inserts cause items to appear on multiple pages):
from pydantic import BaseModel
from typing import Optional
import base64
import json
class CursorPage(BaseModel):
items: list[dict]
next_cursor: Optional[str] = None
prev_cursor: Optional[str] = None
has_more: bool
def encode_cursor(book_id: int, created_at: str) -> str:
"""Encode pagination cursor as base64 JSON."""
data = {"id": book_id, "created_at": created_at}
return base64.urlsafe_b64encode(
json.dumps(data).encode()
).decode()
def decode_cursor(cursor: str) -> dict:
"""Decode a pagination cursor."""
try:
data = json.loads(
base64.urlsafe_b64decode(cursor.encode()).decode()
)
return data
except Exception:
raise HTTPException(
status_code=400, detail="Invalid cursor format"
)
@app.get("/api/v1/books/cursor", response_model=CursorPage)
def list_books_cursor(
cursor: Optional[str] = Query(None, description="Pagination cursor"),
limit: int = Query(20, ge=1, le=100),
):
"""List books with cursor-based pagination."""
all_books = sorted(
book_store.get_all(),
key=lambda b: (b["created_at"], b["id"]),
)
# If cursor provided, find the starting point
start_index = 0
if cursor:
cursor_data = decode_cursor(cursor)
for i, book in enumerate(all_books):
if book["id"] == cursor_data["id"]:
start_index = i + 1
break
# Get one extra item to determine if there are more
items = all_books[start_index : start_index + limit + 1]
has_more = len(items) > limit
items = items[:limit]
# Build cursors
next_cursor = None
if has_more and items:
last = items[-1]
next_cursor = encode_cursor(
last["id"], last["created_at"].isoformat()
)
prev_cursor = None
if start_index > 0 and items:
first = items[0]
prev_cursor = encode_cursor(
first["id"], first["created_at"].isoformat()
)
return CursorPage(
items=items,
next_cursor=next_cursor,
prev_cursor=prev_cursor,
has_more=has_more,
)
Following the RFC 8288 standard, you can include pagination links in response headers:
from fastapi import Response
from math import ceil
@app.get("/api/v1/books")
def list_books(
response: Response,
page: int = Query(1, ge=1),
per_page: int = Query(20, ge=1, le=100),
):
all_books = book_store.get_all()
total = len(all_books)
total_pages = ceil(total / per_page)
start = (page - 1) * per_page
items = all_books[start : start + per_page]
# Build Link header
base_url = "/api/v1/books"
links = []
if page < total_pages:
links.append(
f'<{base_url}?page={page + 1}&per_page={per_page}>; rel="next"'
)
if page > 1:
links.append(
f'<{base_url}?page={page - 1}&per_page={per_page}>; rel="prev"'
)
links.append(
f'<{base_url}?page=1&per_page={per_page}>; rel="first"'
)
links.append(
f'<{base_url}?page={total_pages}&per_page={per_page}>; rel="last"'
)
if links:
response.headers["Link"] = ", ".join(links)
response.headers["X-Total-Count"] = str(total)
response.headers["X-Total-Pages"] = str(total_pages)
return {"items": items, "total": total, "page": page, "per_page": per_page}
| Strategy | Best For | Drawbacks |
|---|---|---|
| Offset/Limit | Simple lists, admin panels, small datasets | Slow on large offsets, shifting window problem |
| Cursor-based | Real-time feeds, large datasets, infinite scroll | Cannot jump to arbitrary page |
| Keyset (ID-based) | Chronological data, activity logs | Requires stable sort order |
Clients need to find specific subsets of data efficiently. Well-designed filtering and sorting makes your API dramatically more useful.
from fastapi import FastAPI, Query
from typing import Optional
from enum import Enum
app = FastAPI()
class SortField(str, Enum):
TITLE = "title"
AUTHOR = "author"
PRICE = "price"
CREATED = "created_at"
YEAR = "published_year"
class SortOrder(str, Enum):
ASC = "asc"
DESC = "desc"
@app.get("/api/v1/books")
def list_books(
# Text search
search: Optional[str] = Query(
None, min_length=2, description="Search in title and author"
),
# Exact match filters
genre: Optional[str] = Query(None, description="Filter by genre"),
author: Optional[str] = Query(None, description="Filter by author name"),
status: Optional[str] = Query(None, description="Filter by status"),
# Range filters
min_price: Optional[float] = Query(None, ge=0, description="Minimum price"),
max_price: Optional[float] = Query(None, ge=0, description="Maximum price"),
published_after: Optional[int] = Query(None, description="Published year minimum"),
published_before: Optional[int] = Query(None, description="Published year maximum"),
# Sorting
sort_by: SortField = Query(SortField.CREATED, description="Sort field"),
sort_order: SortOrder = Query(SortOrder.DESC, description="Sort direction"),
# Pagination
page: int = Query(1, ge=1),
per_page: int = Query(20, ge=1, le=100),
):
books = book_store.get_all()
# Apply text search
if search:
search_lower = search.lower()
books = [
b for b in books
if search_lower in b["title"].lower()
or search_lower in b["author"].lower()
]
# Apply exact match filters
if genre:
books = [b for b in books if b.get("genre", "").lower() == genre.lower()]
if author:
books = [b for b in books if author.lower() in b["author"].lower()]
if status:
books = [b for b in books if b.get("status") == status]
# Apply range filters
if min_price is not None:
books = [b for b in books if b["price"] >= min_price]
if max_price is not None:
books = [b for b in books if b["price"] <= max_price]
if published_after is not None:
books = [
b for b in books
if b.get("published_year") and b["published_year"] >= published_after
]
if published_before is not None:
books = [
b for b in books
if b.get("published_year") and b["published_year"] <= published_before
]
# Apply sorting
reverse = sort_order == SortOrder.DESC
books.sort(
key=lambda b: b.get(sort_by.value, ""),
reverse=reverse,
)
# Apply pagination
total = len(books)
start = (page - 1) * per_page
items = books[start : start + per_page]
return {
"items": items,
"total": total,
"page": page,
"per_page": per_page,
"filters_applied": {
"search": search,
"genre": genre,
"author": author,
"min_price": min_price,
"max_price": max_price,
},
}
Extract filtering logic into a reusable dependency class:
from fastapi import Depends, Query
from dataclasses import dataclass, field
from typing import Optional, Callable
@dataclass
class BookFilters:
"""Reusable book filter parameters."""
search: Optional[str] = Query(None, min_length=2)
genre: Optional[str] = None
author: Optional[str] = None
min_price: Optional[float] = Query(None, ge=0)
max_price: Optional[float] = Query(None, ge=0)
published_after: Optional[int] = None
published_before: Optional[int] = None
in_stock: Optional[bool] = None
def apply(self, books: list[dict]) -> list[dict]:
"""Apply all active filters to a list of books."""
result = books
if self.search:
q = self.search.lower()
result = [
b for b in result
if q in b["title"].lower() or q in b["author"].lower()
]
if self.genre:
result = [
b for b in result
if b.get("genre", "").lower() == self.genre.lower()
]
if self.author:
result = [
b for b in result
if self.author.lower() in b["author"].lower()
]
if self.min_price is not None:
result = [b for b in result if b["price"] >= self.min_price]
if self.max_price is not None:
result = [b for b in result if b["price"] <= self.max_price]
if self.published_after is not None:
result = [
b for b in result
if b.get("published_year")
and b["published_year"] >= self.published_after
]
if self.published_before is not None:
result = [
b for b in result
if b.get("published_year")
and b["published_year"] <= self.published_before
]
return result
@dataclass
class SortParams:
"""Reusable sort parameters."""
sort_by: SortField = Query(SortField.CREATED)
sort_order: SortOrder = Query(SortOrder.DESC)
def apply(self, items: list[dict]) -> list[dict]:
reverse = self.sort_order == SortOrder.DESC
return sorted(
items,
key=lambda x: x.get(self.sort_by.value, ""),
reverse=reverse,
)
# Clean endpoint using dependencies
@app.get("/api/v1/books")
def list_books(
filters: BookFilters = Depends(),
sorting: SortParams = Depends(),
pagination: PaginationParams = Depends(),
):
books = book_store.get_all()
books = filters.apply(books)
books = sorting.apply(books)
total = len(books)
items = books[pagination.offset : pagination.offset + pagination.limit]
return {
"items": items,
"total": total,
"page": pagination.page,
"per_page": pagination.per_page,
}
This pattern keeps endpoints clean and DRY. The same BookFilters and SortParams can be reused across multiple endpoints.
Allow filtering by multiple values for the same field:
from fastapi import Query
from typing import Optional
@app.get("/api/v1/books")
def list_books(
# Multiple genres: ?genres=fiction&genres=science
genres: Optional[list[str]] = Query(None, description="Filter by genres"),
# Multiple authors
authors: Optional[list[str]] = Query(None),
# Multiple statuses
statuses: Optional[list[str]] = Query(None),
# Tags (many-to-many)
tags: Optional[list[str]] = Query(
None, description="Books must have ALL specified tags"
),
any_tags: Optional[list[str]] = Query(
None, description="Books must have ANY of specified tags"
),
):
books = book_store.get_all()
if genres:
genres_lower = [g.lower() for g in genres]
books = [
b for b in books
if b.get("genre", "").lower() in genres_lower
]
if authors:
authors_lower = [a.lower() for a in authors]
books = [
b for b in books
if any(a in b["author"].lower() for a in authors_lower)
]
if statuses:
books = [b for b in books if b.get("status") in statuses]
if tags:
# AND logic — book must have ALL tags
tags_set = set(t.lower() for t in tags)
books = [
b for b in books
if tags_set.issubset(set(t.lower() for t in b.get("tags", [])))
]
if any_tags:
# OR logic — book must have at least one tag
any_tags_set = set(t.lower() for t in any_tags)
books = [
b for b in books
if any_tags_set.intersection(
set(t.lower() for t in b.get("tags", []))
)
]
return {"items": books, "total": len(books)}
As your API evolves, you will need to make breaking changes while still supporting existing clients. API versioning lets you introduce new versions without breaking old integrations.
The most common and explicit approach — include the version in the URL path:
from fastapi import FastAPI, APIRouter
app = FastAPI(title="Versioned API")
# Version 1 router
v1_router = APIRouter(prefix="/api/v1", tags=["v1"])
@v1_router.get("/books")
def list_books_v1():
"""V1: Returns basic book data."""
books = book_store.get_all()
return [
{
"id": b["id"],
"title": b["title"],
"author": b["author"],
"price": b["price"],
}
for b in books
]
@v1_router.get("/books/{book_id}")
def get_book_v1(book_id: int):
book = book_store.get(book_id)
if not book:
raise HTTPException(status_code=404, detail="Book not found")
return {
"id": book["id"],
"title": book["title"],
"author": book["author"],
"price": book["price"],
}
# Version 2 router — enhanced response format
v2_router = APIRouter(prefix="/api/v2", tags=["v2"])
@v2_router.get("/books")
def list_books_v2(
page: int = Query(1, ge=1),
per_page: int = Query(20, ge=1, le=100),
):
"""V2: Returns paginated books with metadata and links."""
books = book_store.get_all()
total = len(books)
start = (page - 1) * per_page
items = books[start : start + per_page]
return {
"data": [
{
"id": b["id"],
"title": b["title"],
"author": {
"name": b["author"],
},
"pricing": {
"amount": b["price"],
"currency": "USD",
},
"metadata": {
"genre": b.get("genre"),
"published_year": b.get("published_year"),
"status": b.get("status", "available"),
},
"links": {
"self": f"/api/v2/books/{b['id']}",
"reviews": f"/api/v2/books/{b['id']}/reviews",
},
}
for b in items
],
"pagination": {
"total": total,
"page": page,
"per_page": per_page,
"total_pages": (total + per_page - 1) // per_page,
},
}
# Register both versions
app.include_router(v1_router)
app.include_router(v2_router)
An alternative approach uses a custom header to specify the API version:
from fastapi import FastAPI, Header, HTTPException
@app.get("/api/books")
def list_books(
api_version: str = Header(
"1", alias="X-API-Version", description="API version"
),
):
"""Route to different implementations based on version header."""
books = book_store.get_all()
if api_version == "1":
return [
{"id": b["id"], "title": b["title"], "author": b["author"]}
for b in books
]
elif api_version == "2":
return {
"data": [
{
"id": b["id"],
"title": b["title"],
"author": {"name": b["author"]},
"pricing": {"amount": b["price"], "currency": "USD"},
}
for b in books
],
"meta": {"version": "2", "total": len(books)},
}
else:
raise HTTPException(
status_code=400,
detail=f"Unsupported API version: {api_version}. Supported: 1, 2",
)
For larger applications, organize versioned code into separate modules:
# Project structure:
# app/
# +-- main.py
# +-- api/
# | +-- __init__.py
# | +-- v1/
# | | +-- __init__.py
# | | +-- router.py
# | | +-- endpoints/
# | | | +-- books.py
# | | | +-- authors.py
# | | +-- schemas.py
# | +-- v2/
# | +-- __init__.py
# | +-- router.py
# | +-- endpoints/
# | | +-- books.py
# | | +-- authors.py
# | +-- schemas.py
# +-- core/
# | +-- config.py
# | +-- database.py
# +-- models/
# +-- book.py
# +-- author.py
# app/api/v1/router.py
from fastapi import APIRouter
from .endpoints import books, authors
router = APIRouter(prefix="/api/v1")
router.include_router(books.router, prefix="/books", tags=["v1-books"])
router.include_router(authors.router, prefix="/authors", tags=["v1-authors"])
# app/api/v2/router.py
from fastapi import APIRouter
from .endpoints import books, authors
router = APIRouter(prefix="/api/v2")
router.include_router(books.router, prefix="/books", tags=["v2-books"])
router.include_router(authors.router, prefix="/authors", tags=["v2-authors"])
# app/main.py
from fastapi import FastAPI
from app.api.v1.router import router as v1_router
from app.api.v2.router import router as v2_router
app = FastAPI(
title="Book Management API",
description="Multi-version REST API",
)
app.include_router(v1_router)
app.include_router(v2_router)
# Deprecation notice for v1
@app.middleware("http")
async def add_deprecation_header(request, call_next):
response = await call_next(request)
if request.url.path.startswith("/api/v1"):
response.headers["Deprecation"] = "true"
response.headers["Sunset"] = "2025-12-31"
response.headers["Link"] = (
'</api/v2>; rel="successor-version"'
)
return response
| Strategy | Pros | Cons |
|---|---|---|
URL path (/v1/books) |
Explicit, easy to test, cacheable | URL pollution, harder to remove versions |
Header (X-API-Version: 2) |
Clean URLs, flexible | Hidden, harder to test, not cacheable |
Query param (?version=2) |
Easy to use, visible | Not standard, pollutes query string |
| Content negotiation | HTTP-standard approach | Complex, poor tooling support |
Some operations should not block the API response — sending emails, processing uploads, generating reports, updating caches. FastAPI provides BackgroundTasks for running code after the response is sent.
from fastapi import FastAPI, BackgroundTasks
from datetime import datetime
import logging
app = FastAPI()
logger = logging.getLogger(__name__)
def send_welcome_email(email: str, username: str):
"""Simulate sending a welcome email (runs in background)."""
logger.info(f"Sending welcome email to {email} for user {username}")
# In production, use an email service like SendGrid, SES, etc.
import time
time.sleep(2) # Simulate slow email API call
logger.info(f"Welcome email sent to {email}")
def log_user_activity(user_id: int, action: str):
"""Log user activity to analytics (runs in background)."""
logger.info(f"Logging activity: user={user_id}, action={action}")
# Write to analytics database, send to event stream, etc.
@app.post("/api/v1/users", status_code=201)
def create_user(user: UserCreate, background_tasks: BackgroundTasks):
"""Create a user and send welcome email in the background."""
# Create user immediately
new_user = user_store.create(user.model_dump())
# Queue background tasks — these run AFTER the response is sent
background_tasks.add_task(
send_welcome_email, new_user["email"], new_user["username"]
)
background_tasks.add_task(
log_user_activity, new_user["id"], "registration"
)
# Response is sent immediately, without waiting for email
return new_user
from fastapi import Depends, BackgroundTasks
import json
from pathlib import Path
class AuditLogger:
"""Service for logging audit events."""
def __init__(self, log_dir: str = "/var/log/api"):
self.log_dir = Path(log_dir)
self.log_dir.mkdir(parents=True, exist_ok=True)
def log_event(
self, event_type: str, resource_type: str,
resource_id: int, user_id: int, details: dict
):
event = {
"timestamp": datetime.utcnow().isoformat(),
"event_type": event_type,
"resource_type": resource_type,
"resource_id": resource_id,
"user_id": user_id,
"details": details,
}
log_file = self.log_dir / f"audit_{datetime.utcnow():%Y%m%d}.jsonl"
with open(log_file, "a") as f:
f.write(json.dumps(event) + "
")
def get_audit_logger():
return AuditLogger()
@app.delete("/api/v1/books/{book_id}", status_code=204)
def delete_book(
book_id: int,
background_tasks: BackgroundTasks,
audit: AuditLogger = Depends(get_audit_logger),
):
book = book_store.get(book_id)
if not book:
raise HTTPException(status_code=404, detail="Book not found")
book_store.delete(book_id)
# Audit logging happens in background
background_tasks.add_task(
audit.log_event,
event_type="delete",
resource_type="book",
resource_id=book_id,
user_id=1, # Would come from auth
details={"title": book["title"], "isbn": book["isbn"]},
)
return Response(status_code=204)
from fastapi import BackgroundTasks
from pydantic import BaseModel
from enum import Enum
import uuid
class ReportStatus(str, Enum):
PENDING = "pending"
PROCESSING = "processing"
COMPLETED = "completed"
FAILED = "failed"
# In-memory report tracker
reports: dict[str, dict] = {}
def generate_report(report_id: str, report_type: str, filters: dict):
"""Generate a report in the background."""
reports[report_id]["status"] = ReportStatus.PROCESSING
try:
# Simulate heavy processing
import time
time.sleep(10)
# Build report data
if report_type == "inventory":
books = book_store.get_all()
report_data = {
"total_books": len(books),
"total_value": sum(b["price"] for b in books),
"by_genre": {},
}
for book in books:
genre = book.get("genre", "unknown")
if genre not in report_data["by_genre"]:
report_data["by_genre"][genre] = 0
report_data["by_genre"][genre] += 1
reports[report_id]["data"] = report_data
reports[report_id]["status"] = ReportStatus.COMPLETED
except Exception as e:
reports[report_id]["status"] = ReportStatus.FAILED
reports[report_id]["error"] = str(e)
class ReportRequest(BaseModel):
report_type: str
filters: dict = {}
@app.post("/api/v1/reports", status_code=202)
def request_report(
request: ReportRequest,
background_tasks: BackgroundTasks,
):
"""Request a report generation. Returns immediately with a report ID."""
report_id = str(uuid.uuid4())
reports[report_id] = {
"id": report_id,
"report_type": request.report_type,
"status": ReportStatus.PENDING,
"created_at": datetime.utcnow().isoformat(),
"data": None,
"error": None,
}
background_tasks.add_task(
generate_report, report_id, request.report_type, request.filters
)
return {
"report_id": report_id,
"status": ReportStatus.PENDING,
"check_status_url": f"/api/v1/reports/{report_id}",
}
@app.get("/api/v1/reports/{report_id}")
def get_report_status(report_id: str):
"""Check the status of a report."""
if report_id not in reports:
raise HTTPException(status_code=404, detail="Report not found")
return reports[report_id]
For production systems with heavy background processing, use Celery with Redis or RabbitMQ as the broker.
Middleware intercepts every request before it reaches your endpoints and every response before it reaches the client. It is the ideal place for cross-cutting concerns like logging, authentication, CORS, rate limiting, and request timing.
from fastapi import FastAPI, Request
from starlette.middleware.base import BaseHTTPMiddleware
import time
import uuid
import logging
app = FastAPI()
logger = logging.getLogger(__name__)
class RequestTimingMiddleware(BaseHTTPMiddleware):
"""Add request timing headers to every response."""
async def dispatch(self, request: Request, call_next):
start_time = time.perf_counter()
response = await call_next(request)
duration = time.perf_counter() - start_time
response.headers["X-Process-Time"] = f"{duration:.4f}"
response.headers["X-Process-Time-Ms"] = f"{duration * 1000:.2f}"
return response
class RequestIDMiddleware(BaseHTTPMiddleware):
"""Add a unique request ID to every request/response."""
async def dispatch(self, request: Request, call_next):
# Use client-provided ID or generate one
request_id = request.headers.get(
"X-Request-ID", str(uuid.uuid4())
)
# Store in request state for use in endpoints
request.state.request_id = request_id
response = await call_next(request)
response.headers["X-Request-ID"] = request_id
return response
class RequestLoggingMiddleware(BaseHTTPMiddleware):
"""Log every request and response."""
async def dispatch(self, request: Request, call_next):
# Log request
logger.info(
f"Request: {request.method} {request.url.path} "
f"client={request.client.host if request.client else 'unknown'}"
)
start_time = time.perf_counter()
response = await call_next(request)
duration = time.perf_counter() - start_time
# Log response
logger.info(
f"Response: {request.method} {request.url.path} "
f"status={response.status_code} duration={duration:.4f}s"
)
return response
# Register middleware (order matters — last added runs first)
app.add_middleware(RequestTimingMiddleware)
app.add_middleware(RequestIDMiddleware)
app.add_middleware(RequestLoggingMiddleware)
Cross-Origin Resource Sharing (CORS) is essential when your API is consumed by web browsers from different domains:
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
app = FastAPI()
# Configure CORS
app.add_middleware(
CORSMiddleware,
# Allowed origins (use specific domains in production)
allow_origins=[
"http://localhost:3000", # React dev server
"http://localhost:5173", # Vite dev server
"https://myapp.example.com", # Production frontend
],
# Allow credentials (cookies, authorization headers)
allow_credentials=True,
# Allowed HTTP methods
allow_methods=["GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"],
# Allowed request headers
allow_headers=[
"Authorization",
"Content-Type",
"X-Request-ID",
"X-API-Version",
],
# Headers exposed to the browser
expose_headers=[
"X-Request-ID",
"X-Process-Time",
"X-Total-Count",
],
# Cache preflight requests for 1 hour
max_age=3600,
)
allow_origins=["*"] with allow_credentials=True in production. This would allow any website to make authenticated requests to your API. Always specify exact allowed origins.from fastapi import Request
from fastapi.responses import JSONResponse
from starlette.middleware.base import BaseHTTPMiddleware
from collections import defaultdict
import time
class RateLimitMiddleware(BaseHTTPMiddleware):
"""Simple in-memory rate limiter."""
def __init__(self, app, requests_per_minute: int = 60):
super().__init__(app)
self.requests_per_minute = requests_per_minute
self.requests: dict[str, list[float]] = defaultdict(list)
async def dispatch(self, request: Request, call_next):
# Get client identifier
client_ip = request.client.host if request.client else "unknown"
# Clean old entries
now = time.time()
window_start = now - 60
self.requests[client_ip] = [
t for t in self.requests[client_ip] if t > window_start
]
# Check rate limit
if len(self.requests[client_ip]) >= self.requests_per_minute:
return JSONResponse(
status_code=429,
content={
"error": "Rate limit exceeded",
"message": f"Maximum {self.requests_per_minute} requests per minute",
"retry_after": 60,
},
headers={
"Retry-After": "60",
"X-RateLimit-Limit": str(self.requests_per_minute),
"X-RateLimit-Remaining": "0",
},
)
# Record request
self.requests[client_ip].append(now)
remaining = self.requests_per_minute - len(self.requests[client_ip])
response = await call_next(request)
# Add rate limit headers
response.headers["X-RateLimit-Limit"] = str(self.requests_per_minute)
response.headers["X-RateLimit-Remaining"] = str(remaining)
response.headers["X-RateLimit-Reset"] = str(int(window_start + 60))
return response
app.add_middleware(RateLimitMiddleware, requests_per_minute=100)
For simple middleware, use the @app.middleware("http") decorator:
from fastapi import FastAPI, Request
app = FastAPI()
@app.middleware("http")
async def add_security_headers(request: Request, call_next):
"""Add security headers to all responses."""
response = await call_next(request)
response.headers["X-Content-Type-Options"] = "nosniff"
response.headers["X-Frame-Options"] = "DENY"
response.headers["X-XSS-Protection"] = "1; mode=block"
response.headers["Strict-Transport-Security"] = (
"max-age=31536000; includeSubDomains"
)
response.headers["Cache-Control"] = "no-store"
return response
@app.middleware("http")
async def strip_trailing_slashes(request: Request, call_next):
"""Normalize URLs by stripping trailing slashes."""
if request.url.path != "/" and request.url.path.endswith("/"):
# Reconstruct URL without trailing slash
from starlette.datastructures import URL
url = request.url.replace(path=request.url.path.rstrip("/"))
request._url = url
return await call_next(request)
FastAPI automatically generates OpenAPI (Swagger) documentation. Customizing it makes your API more discoverable and easier to integrate with. Well-documented APIs dramatically reduce integration time for consumers.
from fastapi import FastAPI
app = FastAPI(
title="Book Management API",
description="""
## Book Management REST API
A comprehensive API for managing books, authors, and reviews.
### Features
* **Books** — Full CRUD with filtering, pagination, and sorting
* **Authors** — Author management with book associations
* **Reviews** — User reviews with ratings and moderation
* **Reports** — Async report generation
### Authentication
All endpoints require a Bearer token in the Authorization header.
See the `/auth/token` endpoint for obtaining tokens.
""",
version="2.1.0",
terms_of_service="https://example.com/terms",
contact={
"name": "API Support",
"url": "https://example.com/support",
"email": "api@example.com",
},
license_info={
"name": "MIT",
"url": "https://opensource.org/licenses/MIT",
},
# Custom docs URLs
docs_url="/docs", # Swagger UI
redoc_url="/redoc", # ReDoc
openapi_url="/openapi.json",
)
from fastapi import FastAPI, APIRouter
# Define tag metadata for documentation
tags_metadata = [
{
"name": "books",
"description": "Operations for managing books in the catalog.",
},
{
"name": "authors",
"description": "Author management and book associations.",
},
{
"name": "reviews",
"description": "Book reviews and ratings by users.",
},
{
"name": "reports",
"description": "Generate and retrieve analytical reports.",
"externalDocs": {
"description": "Report format specification",
"url": "https://example.com/docs/reports",
},
},
{
"name": "admin",
"description": "Administrative operations. **Requires admin role.**",
},
]
app = FastAPI(
title="Book Management API",
openapi_tags=tags_metadata,
)
# Routers with tags
books_router = APIRouter(prefix="/api/v1/books", tags=["books"])
authors_router = APIRouter(prefix="/api/v1/authors", tags=["authors"])
reviews_router = APIRouter(prefix="/api/v1/reviews", tags=["reviews"])
admin_router = APIRouter(prefix="/api/v1/admin", tags=["admin"])
@books_router.get(
"/",
summary="List all books",
description="Retrieve a paginated list of books with optional filtering.",
operation_id="listBooks",
)
def list_books():
return book_store.get_all()
@books_router.post(
"/",
summary="Create a new book",
description="Add a new book to the catalog. ISBN must be unique.",
operation_id="createBook",
response_description="The newly created book",
)
def create_book(book: BookCreate):
return book_store.create(book.model_dump())
app.include_router(books_router)
app.include_router(authors_router)
app.include_router(reviews_router)
app.include_router(admin_router)
from pydantic import BaseModel, Field, ConfigDict
class BookCreate(BaseModel):
title: str = Field(..., min_length=1, max_length=300)
author: str = Field(..., min_length=1, max_length=200)
isbn: str = Field(..., pattern=r"^\d{13}$")
price: float = Field(..., gt=0, le=10000)
genre: str = Field(None, max_length=50)
model_config = ConfigDict(
json_schema_extra={
"examples": [
{
"title": "Clean Code",
"author": "Robert C. Martin",
"isbn": "9780132350884",
"price": 39.99,
"genre": "programming",
},
{
"title": "The Pragmatic Programmer",
"author": "David Thomas, Andrew Hunt",
"isbn": "9780135957059",
"price": 49.99,
"genre": "programming",
},
]
}
)
@books_router.post(
"/",
response_model=BookResponse,
status_code=201,
responses={
201: {
"description": "Book created successfully",
"content": {
"application/json": {
"example": {
"id": 1,
"title": "Clean Code",
"author": "Robert C. Martin",
"isbn": "9780132350884",
"price": 39.99,
"genre": "programming",
"status": "available",
"created_at": "2024-01-15T10:30:00",
}
}
},
},
409: {
"description": "Book with this ISBN already exists",
"content": {
"application/json": {
"example": {
"detail": "Book with ISBN 9780132350884 already exists"
}
}
},
},
},
)
def create_book(book: BookCreate):
return book_store.create(book.model_dump())
@app.get(
"/api/v1/books/search",
deprecated=True,
summary="[DEPRECATED] Search books",
description=(
"**Deprecated**: Use `GET /api/v2/books?search=query` instead. "
"This endpoint will be removed on 2025-06-01."
),
tags=["books"],
)
def search_books_v1(q: str):
"""Old search endpoint — marked as deprecated in docs."""
books = book_store.get_all()
return [b for b in books if q.lower() in b["title"].lower()]
# Hidden from OpenAPI docs but still accessible
@app.get("/internal/health", include_in_schema=False)
def health_check():
return {"status": "healthy"}
@app.get("/internal/metrics", include_in_schema=False)
def metrics():
return {
"total_books": book_store.count(),
"uptime_seconds": time.time() - start_time,
}
Let us bring everything together into a production-ready Book Management API. This project demonstrates all the concepts covered in this tutorial — CRUD operations, pagination, filtering, error handling, background tasks, middleware, and proper project structure.
book_api/ +-- main.py # Application entry point +-- config.py # Configuration settings +-- models.py # Pydantic schemas +-- store.py # Data store (in-memory for demo) +-- routers/ | +-- __init__.py | +-- books.py # Book endpoints | +-- authors.py # Author endpoints | +-- health.py # Health check endpoints +-- middleware/ | +-- __init__.py | +-- timing.py # Request timing | +-- logging_mw.py # Request logging +-- dependencies.py # Shared dependencies +-- exceptions.py # Custom exceptions +-- requirements.txt +-- Dockerfile
# config.py
from pydantic_settings import BaseSettings
from functools import lru_cache
class Settings(BaseSettings):
app_name: str = "Book Management API"
app_version: str = "1.0.0"
debug: bool = False
api_prefix: str = "/api/v1"
# Pagination defaults
default_page_size: int = 20
max_page_size: int = 100
# Rate limiting
rate_limit_per_minute: int = 100
# CORS
allowed_origins: list[str] = ["http://localhost:3000"]
class Config:
env_file = ".env"
@lru_cache
def get_settings() -> Settings:
return Settings()
# exceptions.py
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from datetime import datetime
class AppException(Exception):
def __init__(self, message: str, status_code: int = 500):
self.message = message
self.status_code = status_code
class NotFoundException(AppException):
def __init__(self, resource: str, resource_id):
super().__init__(
message=f"{resource} with ID {resource_id} not found",
status_code=404,
)
class DuplicateException(AppException):
def __init__(self, resource: str, field: str, value: str):
super().__init__(
message=f"{resource} with {field}='{value}' already exists",
status_code=409,
)
class BadRequestException(AppException):
def __init__(self, message: str):
super().__init__(message=message, status_code=400)
def register_exception_handlers(app: FastAPI):
"""Register all custom exception handlers."""
@app.exception_handler(AppException)
async def app_exception_handler(request: Request, exc: AppException):
return JSONResponse(
status_code=exc.status_code,
content={
"success": False,
"error": type(exc).__name__,
"message": exc.message,
"timestamp": datetime.utcnow().isoformat(),
},
)
@app.exception_handler(RequestValidationError)
async def validation_handler(request: Request, exc: RequestValidationError):
errors = []
for error in exc.errors():
field = " -> ".join(str(loc) for loc in error["loc"])
errors.append({"field": field, "message": error["msg"]})
return JSONResponse(
status_code=422,
content={
"success": False,
"error": "ValidationError",
"message": "Request validation failed",
"errors": errors,
},
)
# models.py
from pydantic import BaseModel, Field, ConfigDict, field_validator, computed_field
from typing import Optional, TypeVar, Generic
from datetime import datetime
from enum import Enum
from math import ceil
T = TypeVar("T")
# ---------- Enums ----------
class BookStatus(str, Enum):
AVAILABLE = "available"
CHECKED_OUT = "checked_out"
RESERVED = "reserved"
ARCHIVED = "archived"
class SortField(str, Enum):
TITLE = "title"
AUTHOR = "author"
PRICE = "price"
CREATED = "created_at"
YEAR = "published_year"
class SortOrder(str, Enum):
ASC = "asc"
DESC = "desc"
# ---------- Author Schemas ----------
class AuthorBase(BaseModel):
name: str = Field(..., min_length=1, max_length=200)
bio: Optional[str] = Field(None, max_length=2000)
nationality: Optional[str] = Field(None, max_length=100)
class AuthorCreate(AuthorBase):
pass
class AuthorResponse(AuthorBase):
id: int
book_count: int = 0
created_at: datetime
model_config = ConfigDict(from_attributes=True)
# ---------- Book Schemas ----------
class BookBase(BaseModel):
title: str = Field(..., min_length=1, max_length=300)
author_name: str = Field(..., min_length=1, max_length=200)
isbn: str = Field(..., pattern=r"^\d{10}(\d{3})?$")
genre: Optional[str] = Field(None, max_length=50)
published_year: Optional[int] = Field(None, ge=1000, le=2030)
price: float = Field(..., gt=0, le=10000)
description: Optional[str] = Field(None, max_length=5000)
tags: list[str] = Field(default_factory=list)
@field_validator("tags")
@classmethod
def normalize_tags(cls, v):
return [tag.lower().strip() for tag in v if tag.strip()][:10]
class BookCreate(BookBase):
model_config = ConfigDict(
json_schema_extra={
"examples": [
{
"title": "Clean Code",
"author_name": "Robert C. Martin",
"isbn": "9780132350884",
"genre": "programming",
"published_year": 2008,
"price": 39.99,
"description": "A handbook of agile software craftsmanship",
"tags": ["programming", "best-practices"],
}
]
}
)
class BookUpdate(BaseModel):
title: Optional[str] = Field(None, min_length=1, max_length=300)
author_name: Optional[str] = Field(None, min_length=1, max_length=200)
isbn: Optional[str] = Field(None, pattern=r"^\d{10}(\d{3})?$")
genre: Optional[str] = Field(None, max_length=50)
published_year: Optional[int] = Field(None, ge=1000, le=2030)
price: Optional[float] = Field(None, gt=0, le=10000)
description: Optional[str] = Field(None, max_length=5000)
status: Optional[BookStatus] = None
tags: Optional[list[str]] = None
class BookResponse(BookBase):
id: int
status: BookStatus = BookStatus.AVAILABLE
created_at: datetime
updated_at: Optional[datetime] = None
@computed_field
@property
def is_recent(self) -> bool:
return (datetime.utcnow() - self.created_at).days < 30
model_config = ConfigDict(from_attributes=True)
class BookSummary(BaseModel):
id: int
title: str
author_name: str
price: float
genre: Optional[str] = None
model_config = ConfigDict(from_attributes=True)
# ---------- Pagination ----------
class PaginatedResponse(BaseModel, Generic[T]):
items: list[T]
total: int
page: int
per_page: int
total_pages: int
has_next: bool
has_prev: bool
# ---------- Bulk Operations ----------
class BulkDeleteRequest(BaseModel):
ids: list[int] = Field(..., min_length=1, max_length=100)
class BulkDeleteResponse(BaseModel):
deleted: list[int]
not_found: list[int]
# ---------- Reports ----------
class ReportRequest(BaseModel):
report_type: str = Field(..., pattern=r"^(inventory|sales|popular)$")
filters: dict = Field(default_factory=dict)
# store.py
from datetime import datetime
from typing import Optional
class InMemoryStore:
"""Generic in-memory data store."""
def __init__(self):
self._data: dict[int, dict] = {}
self._next_id: int = 1
def create(self, data: dict) -> dict:
item_id = self._next_id
self._next_id += 1
item = {
"id": item_id,
**data,
"created_at": datetime.utcnow(),
"updated_at": None,
}
self._data[item_id] = item
return item
def get(self, item_id: int) -> Optional[dict]:
return self._data.get(item_id)
def get_all(self) -> list[dict]:
return list(self._data.values())
def update(self, item_id: int, data: dict) -> Optional[dict]:
if item_id not in self._data:
return None
self._data[item_id].update(data)
self._data[item_id]["updated_at"] = datetime.utcnow()
return self._data[item_id]
def delete(self, item_id: int) -> bool:
if item_id not in self._data:
return False
del self._data[item_id]
return True
def exists(self, item_id: int) -> bool:
return item_id in self._data
def count(self) -> int:
return len(self._data)
def find_by(self, field: str, value) -> Optional[dict]:
for item in self._data.values():
if item.get(field) == value:
return item
return None
# Application stores
book_store = InMemoryStore()
author_store = InMemoryStore()
# Seed with sample data
sample_books = [
{
"title": "Clean Code",
"author_name": "Robert C. Martin",
"isbn": "9780132350884",
"genre": "programming",
"published_year": 2008,
"price": 39.99,
"description": "A handbook of agile software craftsmanship",
"tags": ["programming", "best-practices", "clean-code"],
"status": "available",
},
{
"title": "Design Patterns",
"author_name": "Gang of Four",
"isbn": "9780201633610",
"genre": "programming",
"published_year": 1994,
"price": 54.99,
"description": "Elements of Reusable Object-Oriented Software",
"tags": ["programming", "design-patterns", "oop"],
"status": "available",
},
{
"title": "The Pragmatic Programmer",
"author_name": "David Thomas",
"isbn": "9780135957059",
"genre": "programming",
"published_year": 2019,
"price": 49.99,
"description": "Your journey to mastery, 20th Anniversary Edition",
"tags": ["programming", "best-practices", "career"],
"status": "available",
},
{
"title": "Dune",
"author_name": "Frank Herbert",
"isbn": "9780441013593",
"genre": "science-fiction",
"published_year": 1965,
"price": 17.99,
"description": "A stunning blend of adventure and mysticism",
"tags": ["sci-fi", "classic", "space"],
"status": "available",
},
{
"title": "1984",
"author_name": "George Orwell",
"isbn": "9780451524935",
"genre": "dystopian",
"published_year": 1949,
"price": 12.99,
"description": "A dystopian social science fiction novel",
"tags": ["classic", "dystopian", "politics"],
"status": "available",
},
]
for book_data in sample_books:
book_store.create(book_data)
# dependencies.py
from fastapi import Query, Depends
from dataclasses import dataclass
from typing import Optional
from config import Settings, get_settings
from models import SortField, SortOrder
@dataclass
class PaginationParams:
page: int = Query(1, ge=1, description="Page number")
per_page: int = Query(20, ge=1, le=100, description="Items per page")
@property
def offset(self) -> int:
return (self.page - 1) * self.per_page
@property
def limit(self) -> int:
return self.per_page
@dataclass
class BookFilterParams:
search: Optional[str] = Query(None, min_length=2, description="Search title/author")
genre: Optional[str] = Query(None, description="Filter by genre")
author: Optional[str] = Query(None, description="Filter by author")
min_price: Optional[float] = Query(None, ge=0)
max_price: Optional[float] = Query(None, ge=0)
published_after: Optional[int] = Query(None, ge=1000)
published_before: Optional[int] = Query(None, le=2030)
status: Optional[str] = Query(None)
tags: Optional[list[str]] = Query(None)
def apply(self, items: list[dict]) -> list[dict]:
result = items
if self.search:
q = self.search.lower()
result = [
b for b in result
if q in b["title"].lower() or q in b["author_name"].lower()
]
if self.genre:
result = [b for b in result if b.get("genre", "").lower() == self.genre.lower()]
if self.author:
result = [b for b in result if self.author.lower() in b["author_name"].lower()]
if self.min_price is not None:
result = [b for b in result if b["price"] >= self.min_price]
if self.max_price is not None:
result = [b for b in result if b["price"] <= self.max_price]
if self.published_after is not None:
result = [
b for b in result
if b.get("published_year") and b["published_year"] >= self.published_after
]
if self.published_before is not None:
result = [
b for b in result
if b.get("published_year") and b["published_year"] <= self.published_before
]
if self.status:
result = [b for b in result if b.get("status") == self.status]
if self.tags:
tags_set = set(t.lower() for t in self.tags)
result = [
b for b in result
if tags_set.issubset(set(t.lower() for t in b.get("tags", [])))
]
return result
@dataclass
class SortParams:
sort_by: SortField = Query(SortField.CREATED, description="Sort field")
sort_order: SortOrder = Query(SortOrder.DESC, description="Sort order")
def apply(self, items: list[dict]) -> list[dict]:
reverse = self.sort_order == SortOrder.DESC
return sorted(
items,
key=lambda x: x.get(self.sort_by.value, "") or "",
reverse=reverse,
)
# routers/books.py
from fastapi import APIRouter, Depends, Response, BackgroundTasks, status
from math import ceil
from typing import Optional
from models import (
BookCreate, BookUpdate, BookResponse, BookSummary,
PaginatedResponse, BulkDeleteRequest, BulkDeleteResponse,
)
from store import book_store
from dependencies import PaginationParams, BookFilterParams, SortParams
from exceptions import NotFoundException, DuplicateException, BadRequestException
import logging
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/v1/books", tags=["books"])
def log_action(action: str, book_id: int, details: str = ""):
"""Background task for logging book actions."""
logger.info(f"BOOK_ACTION: {action} id={book_id} {details}")
@router.get(
"/",
response_model=PaginatedResponse[BookSummary],
summary="List books",
description="Retrieve a paginated, filterable, sortable list of books.",
)
def list_books(
filters: BookFilterParams = Depends(),
sorting: SortParams = Depends(),
pagination: PaginationParams = Depends(),
):
books = book_store.get_all()
books = filters.apply(books)
books = sorting.apply(books)
total = len(books)
items = books[pagination.offset : pagination.offset + pagination.limit]
total_pages = ceil(total / pagination.per_page) if pagination.per_page else 0
return {
"items": items,
"total": total,
"page": pagination.page,
"per_page": pagination.per_page,
"total_pages": total_pages,
"has_next": pagination.page < total_pages,
"has_prev": pagination.page > 1,
}
@router.get(
"/{book_id}",
response_model=BookResponse,
summary="Get a book",
responses={404: {"description": "Book not found"}},
)
def get_book(book_id: int):
book = book_store.get(book_id)
if not book:
raise NotFoundException("Book", book_id)
return book
@router.post(
"/",
response_model=BookResponse,
status_code=status.HTTP_201_CREATED,
summary="Create a book",
)
def create_book(
book: BookCreate,
background_tasks: BackgroundTasks,
):
existing = book_store.find_by("isbn", book.isbn)
if existing:
raise DuplicateException("Book", "isbn", book.isbn)
created = book_store.create({**book.model_dump(), "status": "available"})
background_tasks.add_task(log_action, "CREATE", created["id"], book.title)
return created
@router.put(
"/{book_id}",
response_model=BookResponse,
summary="Replace a book",
)
def replace_book(
book_id: int,
book: BookCreate,
background_tasks: BackgroundTasks,
):
if not book_store.exists(book_id):
raise NotFoundException("Book", book_id)
existing = book_store.find_by("isbn", book.isbn)
if existing and existing["id"] != book_id:
raise DuplicateException("Book", "isbn", book.isbn)
updated = book_store.update(book_id, book.model_dump())
background_tasks.add_task(log_action, "REPLACE", book_id)
return updated
@router.patch(
"/{book_id}",
response_model=BookResponse,
summary="Update a book partially",
)
def update_book(
book_id: int,
book: BookUpdate,
background_tasks: BackgroundTasks,
):
if not book_store.exists(book_id):
raise NotFoundException("Book", book_id)
update_data = book.model_dump(exclude_unset=True)
if not update_data:
raise BadRequestException("No fields provided for update")
if "isbn" in update_data:
existing = book_store.find_by("isbn", update_data["isbn"])
if existing and existing["id"] != book_id:
raise DuplicateException("Book", "isbn", update_data["isbn"])
updated = book_store.update(book_id, update_data)
background_tasks.add_task(log_action, "UPDATE", book_id, str(update_data.keys()))
return updated
@router.delete(
"/{book_id}",
status_code=status.HTTP_204_NO_CONTENT,
summary="Delete a book",
)
def delete_book(book_id: int, background_tasks: BackgroundTasks):
if not book_store.exists(book_id):
raise NotFoundException("Book", book_id)
book_store.delete(book_id)
background_tasks.add_task(log_action, "DELETE", book_id)
return Response(status_code=status.HTTP_204_NO_CONTENT)
@router.post(
"/bulk-delete",
response_model=BulkDeleteResponse,
summary="Delete multiple books",
)
def bulk_delete(request: BulkDeleteRequest, background_tasks: BackgroundTasks):
deleted = []
not_found = []
for book_id in request.ids:
if book_store.delete(book_id):
deleted.append(book_id)
else:
not_found.append(book_id)
background_tasks.add_task(
log_action, "BULK_DELETE", 0, f"deleted={deleted}"
)
return {"deleted": deleted, "not_found": not_found}
# main.py
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from starlette.middleware.base import BaseHTTPMiddleware
from contextlib import asynccontextmanager
import time
import uuid
import logging
from config import get_settings
from exceptions import register_exception_handlers
from routers.books import router as books_router
# Configure logging
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
)
logger = logging.getLogger(__name__)
# Lifespan event handler
@asynccontextmanager
async def lifespan(app: FastAPI):
"""Startup and shutdown events."""
settings = get_settings()
logger.info(f"Starting {settings.app_name} v{settings.app_version}")
logger.info(f"Debug mode: {settings.debug}")
yield # Application runs here
logger.info("Shutting down application")
# Create application
settings = get_settings()
app = FastAPI(
title=settings.app_name,
version=settings.app_version,
description="A comprehensive Book Management REST API built with FastAPI.",
lifespan=lifespan,
docs_url="/docs",
redoc_url="/redoc",
)
# --- Middleware ---
class TimingMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request, call_next):
start = time.perf_counter()
response = await call_next(request)
duration = time.perf_counter() - start
response.headers["X-Process-Time"] = f"{duration:.4f}"
return response
class RequestIDMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request, call_next):
request_id = request.headers.get("X-Request-ID", str(uuid.uuid4()))
request.state.request_id = request_id
response = await call_next(request)
response.headers["X-Request-ID"] = request_id
return response
app.add_middleware(TimingMiddleware)
app.add_middleware(RequestIDMiddleware)
app.add_middleware(
CORSMiddleware,
allow_origins=settings.allowed_origins,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# --- Exception Handlers ---
register_exception_handlers(app)
# --- Routers ---
app.include_router(books_router)
# --- Health Check ---
@app.get("/health", tags=["health"])
def health_check():
return {
"status": "healthy",
"version": settings.app_version,
"timestamp": time.time(),
}
@app.get("/", tags=["root"])
def root():
return {
"name": settings.app_name,
"version": settings.app_version,
"docs": "/docs",
"redoc": "/redoc",
}
# Run with: uvicorn main:app --reload
# Dockerfile FROM python:3.12-slim WORKDIR /app # Install dependencies COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt # Copy application code COPY . . # Create non-root user RUN adduser --disabled-password --gecos "" appuser USER appuser # Expose port EXPOSE 8000 # Run with uvicorn CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]
# requirements.txt fastapi==0.115.0 uvicorn[standard]==0.30.0 pydantic[email]==2.9.0 pydantic-settings==2.5.0 python-multipart==0.0.9
Once the application is running (uvicorn main:app --reload), test it with curl:
# List all books (paginated)
curl -s http://localhost:8000/api/v1/books | python -m json.tool
# Get a specific book
curl -s http://localhost:8000/api/v1/books/1 | python -m json.tool
# Create a new book
curl -s -X POST http://localhost:8000/api/v1/books -H "Content-Type: application/json" -d '{
"title": "Fluent Python",
"author_name": "Luciano Ramalho",
"isbn": "9781492056355",
"genre": "programming",
"published_year": 2022,
"price": 59.99,
"description": "Clear, concise, and effective programming",
"tags": ["python", "advanced"]
}' | python -m json.tool
# Search books
curl -s "http://localhost:8000/api/v1/books?search=clean&sort_by=price&sort_order=asc" | python -m json.tool
# Filter by genre and price range
curl -s "http://localhost:8000/api/v1/books?genre=programming&min_price=30&max_price=50" | python -m json.tool
# Partial update
curl -s -X PATCH http://localhost:8000/api/v1/books/1 -H "Content-Type: application/json" -d '{"price": 34.99, "status": "reserved"}' | python -m json.tool
# Delete a book
curl -s -X DELETE http://localhost:8000/api/v1/books/4
# Bulk delete
curl -s -X POST http://localhost:8000/api/v1/books/bulk-delete -H "Content-Type: application/json" -d '{"ids": [3, 4, 99]}' | python -m json.tool
# Health check
curl -s http://localhost:8000/health | python -m json.tool
Visit http://localhost:8000/docs to see the interactive Swagger UI documentation, or http://localhost:8000/redoc for the ReDoc view.
Building a production-quality REST API with FastAPI requires understanding not just the framework, but REST architecture itself. Here is a summary of the most important lessons from this tutorial:
/books not /getBooks/api/v1/ than to add it laterresponse_model to filter passwords, API keys, and internal fieldsexclude_unset=True for PATCH endpoints to distinguish between “not provided” and “set to null”responses parameter?genres=fiction&genres=science) for flexible queryingallow_origins=["*"] with credentials in production| Operation | Method | Path | Status Code | Returns |
|---|---|---|---|---|
| List (paginated) | GET | /api/v1/books |
200 | Paginated list |
| Get single | GET | /api/v1/books/{id} |
200 | Single resource |
| Create | POST | /api/v1/books |
201 | Created resource |
| Full replace | PUT | /api/v1/books/{id} |
200 | Updated resource |
| Partial update | PATCH | /api/v1/books/{id} |
200 | Updated resource |
| Delete | DELETE | /api/v1/books/{id} |
204 | No content |
| Bulk create | POST | /api/v1/books/bulk |
201 | Created list + errors |
| Bulk delete | POST | /api/v1/books/bulk-delete |
200 | Deleted IDs + not found |
| Search | GET | /api/v1/books?search=term |
200 | Filtered list |
| Async task | POST | /api/v1/reports |
202 | Task ID + status URL |
With these patterns and the complete Book Management API project as a reference, you have everything you need to build robust, well-structured REST APIs with FastAPI. In the next tutorials, we will cover database integration with SQLAlchemy, testing strategies, and authentication and authorization.