What Is FastAPI?
FastAPI is a modern web framework for building high-performance REST APIs quickly with Python. Based on Python type hints, it provides automatic validation, automatic API documentation (Swagger/OpenAPI), and native support for async (async/await).
It is rapidly gaining popularity for being faster than Flask, more concise than Django REST Framework, and delivering performance on par with Node.js and Go. This article covers everything from installation to CRUD API implementation, dependency injection, and error handling.
Installation and Your First API
pip install fastapi uvicorn[standard]
# fastapi: Web framework
# uvicorn: ASGI server (async server)
# main.py
from fastapi import FastAPI
app = FastAPI(
title="Blog API",
description="Blog post management API",
version="1.0.0",
)
@app.get("/")
async def root():
"""API health check"""
return {"message": "Blog API is running", "status": "ok"}
@app.get("/hello/{name}")
async def hello(name: str):
"""Greets the given name."""
return {"message": f"Hello, {name}!"}
# Start the server
uvicorn main:app --reload --port 8000
# Check auto-generated API docs
# Swagger UI: http://localhost:8000/docs
# ReDoc: http://localhost:8000/redoc
Once the server is running, Swagger UI documentation is automatically generated based on your type hints. You can test and share APIs without any separate documentation effort.
Pydantic Models — Request/Response Schemas
FastAPI uses Pydantic models to automatically validate request bodies. When invalid data is received, it returns a 422 error with detailed error messages.
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from datetime import datetime
app = FastAPI()
# Pydantic model — define request/response schemas
class PostCreate(BaseModel):
"""Post creation request"""
title: str = Field(min_length=1, max_length=200,
description="Post title")
content: str = Field(min_length=10,
description="Post body (10+ characters)")
tags: list[str] = Field(default_factory=list,
description="List of tags")
class PostResponse(BaseModel):
"""Post response"""
id: int
title: str
content: str
tags: list[str]
created_at: datetime
# In-memory DB (for demonstration)
posts_db: dict[int, dict] = {}
next_id = 1
@app.post("/posts", response_model=PostResponse,
status_code=201)
async def create_post(post: PostCreate):
"""Creates a new post."""
global next_id
now = datetime.now()
new_post = {
"id": next_id,
"title": post.title,
"content": post.content,
"tags": post.tags,
"created_at": now,
}
posts_db[next_id] = new_post
next_id += 1
return new_post
@app.get("/posts/{post_id}", response_model=PostResponse)
async def get_post(post_id: int):
"""Retrieves a post."""
if post_id not in posts_db:
raise HTTPException(status_code=404,
detail="Post not found")
return posts_db[post_id]
If the constraints defined in the PostCreate model (min_length, max_length) are violated, FastAPI automatically returns a 422 error. There is no need to write validation code yourself.
Query Parameters and Pagination
from fastapi import Query
@app.get("/posts", response_model=list[PostResponse])
async def list_posts(
skip: int = Query(default=0, ge=0,
description="Number of items to skip"),
limit: int = Query(default=10, ge=1, le=100,
description="Number of items to fetch"),
tag: str | None = Query(default=None,
description="Filter by tag"),
):
"""Lists posts with pagination support."""
all_posts = list(posts_db.values())
# Tag filtering
if tag:
all_posts = [p for p in all_posts if tag in p["tags"]]
# Apply pagination
return all_posts[skip : skip + limit]
# GET /posts?skip=0&limit=5&tag=python
Using Query lets you declaratively define parameter defaults, range constraints, and descriptions, all of which are automatically reflected in the Swagger documentation.
Dependency Injection
FastAPI’s dependency injection system cleanly separates cross-cutting concerns like authentication and DB session management.
from fastapi import Depends, Header, HTTPException
# Authentication dependency
async def verify_api_key(
x_api_key: str = Header(description="API authentication key")
) -> str:
"""Verifies the API key."""
valid_keys = {"key-abc-123", "key-xyz-789"}
if x_api_key not in valid_keys:
raise HTTPException(
status_code=401,
detail="Invalid API key"
)
return x_api_key
# Protected endpoint — inject dependency with Depends
@app.delete("/posts/{post_id}")
async def delete_post(
post_id: int,
api_key: str = Depends(verify_api_key), # Authentication required
):
"""Deletes a post (authentication required)."""
if post_id not in posts_db:
raise HTTPException(status_code=404,
detail="Post not found")
deleted = posts_db.pop(post_id)
return {"message": f"Post '{deleted['title']}' deleted"}
# DB session dependency example
async def get_db_session():
"""Creates and closes a DB session."""
session = {"connected": True} # Actual DB connection in practice
try:
yield session # Provide session via yield
finally:
session["connected"] = False # Cleanup after request completes
Using Depends separates authentication logic from each endpoint, eliminating code duplication and making it easy to swap dependencies during testing.
Error Handling
from fastapi import Request
from fastapi.responses import JSONResponse
class PostNotFoundError(Exception):
"""Exception raised when a post cannot be found"""
def __init__(self, post_id: int):
self.post_id = post_id
# Register custom exception handler
@app.exception_handler(PostNotFoundError)
async def post_not_found_handler(
request: Request, exc: PostNotFoundError
):
return JSONResponse(
status_code=404,
content={
"error": "POST_NOT_FOUND",
"message": f"Post with ID {exc.post_id} not found",
"path": str(request.url),
},
)
# Usage example
@app.get("/posts/{post_id}/detail")
async def get_post_detail(post_id: int):
if post_id not in posts_db:
raise PostNotFoundError(post_id)
return posts_db[post_id]
# Response example:
# {
# "error": "POST_NOT_FOUND",
# "message": "Post with ID 99 not found",
# "path": "http://localhost:8000/posts/99/detail"
# }
Middleware — Common Request/Response Processing
import time
from fastapi.middleware.cors import CORSMiddleware
# CORS configuration
app.add_middleware(
CORSMiddleware,
allow_origins=["http://localhost:3000"], # Frontend domain
allow_methods=["*"],
allow_headers=["*"],
)
# Custom middleware: response time measurement
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
start = time.perf_counter()
response = await call_next(request)
elapsed = time.perf_counter() - start
response.headers["X-Process-Time"] = f"{elapsed:.4f}"
return response
Practical Tips
- Project structure: Separate endpoints by module using routers (
APIRouter) - Environment variables: Manage configuration values type-safely with
pydantic-settings - Async DB: Use
sqlalchemy[asyncio]ortortoise-ormfor async database access - Testing: Write async tests with
httpx.AsyncClient+pytest-asyncio - Deployment: Use
gunicorn -k uvicorn.workers.UvicornWorkerinstead of plainuvicornfor production - Versioning: Manage API versions with URL prefixes (
/api/v1/) - Response models: Always specify
response_modelto make response schemas explicit