Building RESTful APIs with FastAPI: Best Practices and Patterns
Introduction
FastAPI has become one of the most popular Python web frameworks for building APIs, thanks to its speed, automatic documentation, and type safety. However, building production-ready APIs requires more than just basic CRUD operations. In this guide, we'll explore best practices and patterns for creating robust, scalable, and maintainable FastAPI applications.
🚀 Why FastAPI? FastAPI offers automatic OpenAPI documentation, type validation with Pydantic, async support, and performance comparable to Node.js and Go.
Project Structure
Organizing Your FastAPI Application
A well-organized project structure is crucial for maintainability:
my-api/
├── app/
│ ├── __init__.py
│ ├── main.py
│ ├── config.py
│ ├── dependencies.py
│ ├── models/
│ │ ├── __init__.py
│ │ ├── user.py
│ │ └── product.py
│ ├── schemas/
│ │ ├── __init__.py
│ │ ├── user.py
│ │ └── product.py
│ ├── api/
│ │ ├── __init__.py
│ │ ├── v1/
│ │ │ ├── __init__.py
│ │ │ ├── router.py
│ │ │ ├── endpoints/
│ │ │ │ ├── __init__.py
│ │ │ │ ├── users.py
│ │ │ │ └── products.py
│ ├── core/
│ │ ├── __init__.py
│ │ ├── security.py
│ │ └── config.py
│ ├── db/
│ │ ├── __init__.py
│ │ ├── base.py
│ │ └── session.py
│ └── utils/
│ ├── __init__.py
│ └── helpers.py
├── tests/
├── requirements.txt
└── README.md
Configuration Management
Environment-Based Configuration
Use Pydantic Settings for type-safe configuration:
# app/core/config.py
from pydantic_settings import BaseSettings
from typing import Optional
class Settings(BaseSettings):
# App settings
APP_NAME: str = "My API"
DEBUG: bool = False
VERSION: str = "1.0.0"
# Database settings
DATABASE_URL: str
DB_POOL_SIZE: int = 10
DB_MAX_OVERFLOW: int = 20
# Security settings
SECRET_KEY: str
ALGORITHM: str = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
# CORS settings
CORS_ORIGINS: list[str] = ["http://localhost:3000"]
# Redis settings (optional)
REDIS_URL: Optional[str] = None
class Config:
env_file = ".env"
case_sensitive = True
settings = Settings()
Database Setup with SQLAlchemy
Async Database Session
Set up async database sessions properly:
# app/db/session.py
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from sqlalchemy.orm import declarative_base
from app.core.config import settings
engine = create_async_engine(
settings.DATABASE_URL,
echo=settings.DEBUG,
pool_size=settings.DB_POOL_SIZE,
max_overflow=settings.DB_MAX_OVERFLOW,
pool_pre_ping=True,
)
AsyncSessionLocal = async_sessionmaker(
engine,
class_=AsyncSession,
expire_on_commit=False,
autocommit=False,
autoflush=False,
)
Base = declarative_base()
# Dependency for getting database session
async def get_db() -> AsyncSession:
async with AsyncSessionLocal() as session:
try:
yield session
await session.commit()
except Exception:
await session.rollback()
raise
finally:
await session.close()
Database Models
Define models with proper relationships:
# app/models/user.py
from sqlalchemy import Column, Integer, String, Boolean, DateTime
from sqlalchemy.orm import relationship
from sqlalchemy.sql import func
from app.db.base import Base
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
email = Column(String, unique=True, index=True, nullable=False)
hashed_password = Column(String, nullable=False)
full_name = Column(String, nullable=True)
is_active = Column(Boolean, default=True)
is_superuser = Column(Boolean, default=False)
created_at = Column(DateTime(timezone=True), server_default=func.now())
updated_at = Column(DateTime(timezone=True), onupdate=func.now())
# Relationships
products = relationship("Product", back_populates="owner")
Pydantic Schemas
Request and Response Schemas
Use separate schemas for requests and responses:
# app/schemas/user.py
from pydantic import BaseModel, EmailStr, ConfigDict
from datetime import datetime
from typing import Optional
# Base schema with common fields
class UserBase(BaseModel):
email: EmailStr
full_name: Optional[str] = None
is_active: bool = True
# Schema for creating a user
class UserCreate(UserBase):
password: str
# Schema for updating a user
class UserUpdate(BaseModel):
email: Optional[EmailStr] = None
full_name: Optional[str] = None
is_active: Optional[bool] = None
# Schema for user response
class UserResponse(UserBase):
id: int
created_at: datetime
updated_at: Optional[datetime] = None
model_config = ConfigDict(from_attributes=True)
# Schema for user in list (without sensitive data)
class UserInList(BaseModel):
id: int
email: EmailStr
full_name: Optional[str] = None
model_config = ConfigDict(from_attributes=True)
API Endpoints
RESTful Endpoint Structure
Follow RESTful conventions:
# app/api/v1/endpoints/users.py
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from typing import List
from app.db.session import get_db
from app.schemas.user import UserCreate, UserUpdate, UserResponse, UserInList
from app.models.user import User
from app.core.dependencies import get_current_user
from sqlalchemy import select
router = APIRouter(prefix="/users", tags=["users"])
@router.post(
"/",
response_model=UserResponse,
status_code=status.HTTP_201_CREATED,
summary="Create a new user",
description="Create a new user with email and password"
)
async def create_user(
user_data: UserCreate,
db: AsyncSession = Depends(get_db)
):
"""Create a new user."""
# Check if user exists
result = await db.execute(
select(User).where(User.email == user_data.email)
)
existing_user = result.scalar_one_or_none()
if existing_user:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="User with this email already exists"
)
# Create user (hash password in a service layer)
new_user = User(**user_data.model_dump())
db.add(new_user)
await db.commit()
await db.refresh(new_user)
return new_user
@router.get(
"/",
response_model=List[UserInList],
summary="List all users",
description="Get a list of all users"
)
async def list_users(
skip: int = 0,
limit: int = 100,
db: AsyncSession = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""List all users with pagination."""
result = await db.execute(
select(User).offset(skip).limit(limit)
)
users = result.scalars().all()
return users
@router.get(
"/{user_id}",
response_model=UserResponse,
summary="Get user by ID",
description="Retrieve a specific user by their ID"
)
async def get_user(
user_id: int,
db: AsyncSession = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Get a specific user by ID."""
result = await db.execute(
select(User).where(User.id == user_id)
)
user = result.scalar_one_or_none()
if not user:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="User not found"
)
return user
@router.patch(
"/{user_id}",
response_model=UserResponse,
summary="Update user",
description="Update user information"
)
async def update_user(
user_id: int,
user_data: UserUpdate,
db: AsyncSession = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Update a user."""
result = await db.execute(
select(User).where(User.id == user_id)
)
user = result.scalar_one_or_none()
if not user:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="User not found"
)
# Update only provided fields
update_data = user_data.model_dump(exclude_unset=True)
for field, value in update_data.items():
setattr(user, field, value)
await db.commit()
await db.refresh(user)
return user
@router.delete(
"/{user_id}",
status_code=status.HTTP_204_NO_CONTENT,
summary="Delete user",
description="Delete a user by ID"
)
async def delete_user(
user_id: int,
db: AsyncSession = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Delete a user."""
result = await db.execute(
select(User).where(User.id == user_id)
)
user = result.scalar_one_or_none()
if not user:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="User not found"
)
await db.delete(user)
await db.commit()
return None
Error Handling
Custom Exception Handlers
Create custom exception handlers for better error responses:
# app/core/exceptions.py
from fastapi import Request, status
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException
class AppException(Exception):
def __init__(self, message: str, status_code: int = 500):
self.message = message
self.status_code = status_code
# app/main.py
from fastapi import FastAPI, Request
from fastapi.exceptions import RequestValidationError
from app.core.exceptions import AppException
app = FastAPI()
@app.exception_handler(AppException)
async def app_exception_handler(request: Request, exc: AppException):
return JSONResponse(
status_code=exc.status_code,
content={
"error": True,
"message": exc.message,
"path": request.url.path
}
)
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
return JSONResponse(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
content={
"error": True,
"message": "Validation error",
"details": exc.errors()
}
)
@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request: Request, exc: StarletteHTTPException):
return JSONResponse(
status_code=exc.status_code,
content={
"error": True,
"message": exc.detail,
"path": request.url.path
}
)
Authentication and Authorization
JWT Authentication
Implement JWT authentication:
# app/core/security.py
from datetime import datetime, timedelta
from typing import Optional
from jose import JWTError, jwt
from passlib.context import CryptContext
from app.core.config import settings
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
def verify_password(plain_password: str, hashed_password: str) -> bool:
return pwd_context.verify(plain_password, hashed_password)
def get_password_hash(password: str) -> str:
return pwd_context.hash(password)
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
to_encode = data.copy()
if expires_delta:
expire = datetime.utcnow() + expires_delta
else:
expire = datetime.utcnow() + timedelta(
minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES
)
to_encode.update({"exp": expire})
encoded_jwt = jwt.encode(
to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM
)
return encoded_jwt
def decode_access_token(token: str) -> Optional[dict]:
try:
payload = jwt.decode(
token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM]
)
return payload
except JWTError:
return None
Dependency for Current User
# app/core/dependencies.py
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.ext.asyncio import AsyncSession
from app.db.session import get_db
from app.models.user import User
from app.core.security import decode_access_token
from sqlalchemy import select
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="api/v1/auth/login")
async def get_current_user(
token: str = Depends(oauth2_scheme),
db: AsyncSession = Depends(get_db)
) -> User:
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
payload = decode_access_token(token)
if payload is None:
raise credentials_exception
user_id: int = payload.get("sub")
if user_id is None:
raise credentials_exception
result = await db.execute(select(User).where(User.id == user_id))
user = result.scalar_one_or_none()
if user is None:
raise credentials_exception
return user
API Versioning
Version Your API
Implement API versioning:
# app/api/v1/router.py
from fastapi import APIRouter
from app.api.v1.endpoints import users, products, auth
api_router = APIRouter()
api_router.include_router(auth.router, prefix="/auth", tags=["authentication"])
api_router.include_router(users.router, prefix="/users", tags=["users"])
api_router.include_router(products.router, prefix="/products", tags=["products"])
# app/main.py
from fastapi import FastAPI
from app.api.v1.router import api_router
app = FastAPI(
title="My API",
description="A comprehensive FastAPI application",
version="1.0.0"
)
app.include_router(api_router, prefix="/api/v1")
Testing
Writing Tests
Create comprehensive tests:
# tests/test_users.py
import pytest
from httpx import AsyncClient
from app.main import app
@pytest.mark.asyncio
async def test_create_user():
async with AsyncClient(app=app, base_url="http://test") as client:
response = await client.post(
"/api/v1/users/",
json={
"email": "test@example.com",
"password": "testpassword123",
"full_name": "Test User"
}
)
assert response.status_code == 201
data = response.json()
assert data["email"] == "test@example.com"
assert "id" in data
@pytest.mark.asyncio
async def test_get_user_not_found():
async with AsyncClient(app=app, base_url="http://test") as client:
response = await client.get("/api/v1/users/999")
assert response.status_code == 404
Documentation
Enhancing OpenAPI Documentation
Customize your API documentation:
# app/main.py
from fastapi import FastAPI
from fastapi.openapi.utils import get_openapi
app = FastAPI(
title="My API",
description="""
A comprehensive REST API built with FastAPI.
## Features
* User authentication
* Product management
* Real-time updates
""",
version="1.0.0",
contact={
"name": "API Support",
"email": "support@example.com",
},
license_info={
"name": "MIT",
},
)
def custom_openapi():
if app.openapi_schema:
return app.openapi_schema
openapi_schema = get_openapi(
title="My API",
version="1.0.0",
description="Custom API documentation",
routes=app.routes,
)
# Add custom security schemes
openapi_schema["components"]["securitySchemes"] = {
"Bearer": {
"type": "http",
"scheme": "bearer",
"bearerFormat": "JWT",
}
}
app.openapi_schema = openapi_schema
return app.openapi_schema
app.openapi = custom_openapi
Conclusion
Building production-ready FastAPI applications requires attention to structure, error handling, security, and documentation. By following these best practices, you'll create APIs that are maintainable, scalable, and secure.
🚀 Next Steps: Consider implementing rate limiting, caching with Redis, background tasks, and monitoring with tools like Prometheus.
Resources:
Related Articles: