Building a Robust FastAPI Application: From CRUD to Docker

Overview
FastAPI is a modern, high-performance web framework for building APIs with Python. Its key features speed, automatic interactive documentation, and simple, elegant syntax make it a top choice for developers.
In this guide, we'll walk through a complete FastAPI application that goes beyond a simple "Hello, World." We'll build an API with CRUD functionality, robust data validation, custom exceptions, JWT authentication, rate limiting, and a Dockerfile for easy deployment.
Let's get started! ๐
๐๏ธ Project Structure
A well-organized project is crucial for maintainability. Our application is structured into logical modules for clarity and separation of concerns.
/
โโโ app/
โ โโโ __init__.py
โ โโโ main.py # App entry point & configuration
โ โโโ crud.py # Business logic for CRUD
โ โโโ database.py # In-memory database simulation
โ โโโ exceptions.py # Custom exception handlers
โ โโโ middleware.py # Rate limiting middleware
โ โโโ models.py # Pydantic data models
โ โโโ security.py # JWT and password handling
โ โโโ routers/
โ โโโ auth.py # Authentication endpoints
โ โโโ items.py # Item-related CRUD endpoints
โโโ Dockerfile # Containerization script
โโโ requirements.txt # Project dependencies
๐ ๏ธ Setup and Installation
First, let's get the project running locally.
Create and activate a virtual environment to isolate our project's dependencies.
python -m venv .venv # On macOS/Linux source .venv/bin/activate # On Windows (PowerShell) .\.venv\scripts\activate.ps1Install the required packages from the
requirements.txtfile.pip install -r requirements.txt
๐งฉ The Building Blocks of Our Application
Let's break down each component of our application to understand how it all fits together.
1. Pydantic Models (app/models.py)
Pydantic is at the core of FastAPI's power. It enforces type hints at runtime, providing automatic data validation, serialization, and documentation. We define the "shape" of our API data here.
# app/models.py
from pydantic import BaseModel
# For creating new items
class ItemCreate(BaseModel):
name: str
description: str | None = None
# For responses, includes the generated ID
class ItemResponse(ItemCreate):
id: int
# For user registration and login
class UserCreate(BaseModel):
username: str
password: str
# For the login response token
class TokenResponse(BaseModel):
access_token: str
2. In-Memory Database (app/database.py)
To keep this example simple and self-contained, we're using a Python dictionary as an in-memory database. This is great for prototyping and testing without needing a real database server.
# app/database.py
from typing import Dict
from .models import ItemResponse, ItemCreate
# In-memory store
_items: Dict[int, ItemResponse] = {}
_next_id = 1
def create_item(data: ItemCreate) -> ItemResponse:
global _next_id
item = ItemResponse(id=_next_id, **data.model_dump())
_items[_next_id] = item
_next_id += 1
return item
def get_item(item_id: int) -> ItemResponse | None:
return _items.get(item_id)
# ... other database functions: update_item, delete_item, list_items
3. CRUD Logic (app/crud.py)
This module separates our business logic from the API routing. It acts as a bridge between our endpoints and the database, handling the core Create, Read, Update, and Delete operations and raising exceptions when necessary.
# app/crud.py
from .models import ItemCreate, ItemResponse
from .exceptions import ItemNotFoundException
from . import database
def create_item(data: ItemCreate) -> ItemResponse:
return database.create_item(data)
def read_item(item_id: int) -> ItemResponse:
item = database.get_item(item_id)
if not item:
raise ItemNotFoundException(item_id)
return item
# ... other CRUD functions: update_item, delete_item, list_items
4. Custom Exceptions (app/exceptions.py)
Clear error handling makes for a better API. We define a custom exception, ItemNotFoundException, and a handler that returns a clean 404 Not Found response instead of a generic server error.
# app/exceptions.py
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
class ItemNotFoundException(Exception):
def __init__(self, item_id: int):
self.item_id = item_id
def register_exception_handlers(app: FastAPI):
@app.exception_handler(ItemNotFoundException)
async def item_not_found_handler(request: Request, exc: ItemNotFoundException):
return JSONResponse(
status_code=404,
content={"detail": f"Item with ID {exc.item_id} not found"},
)
5. Authentication and Security (routers/auth.py and security.py)
We secure our endpoints using JWT (JSON Web Tokens). The auth.py router handles user registration and login, while a security.py module manages password hashing and token generation.
# app/routers/auth.py
from fastapi import APIRouter, HTTPException
from ..models import UserCreate, TokenResponse
from ..security import hash_password, verify_password, create_access_token
router = APIRouter()
_users: dict[str, str] = {} # In-memory user store
@router.post("/register", status_code=200)
def register(user: UserCreate):
if user.username in _users:
raise HTTPException(status_code=400, detail="Username already exists")
_users[user.username] = hash_password(user.password)
return {"message": "User registered"}
@router.post("/login", response_model=TokenResponse)
def login(user: UserCreate):
# ... login logic ...
token = create_access_token(subject=user.username)
return TokenResponse(access_token=token)
# app/security.py
from datetime import datetime, timedelta, timezone
from typing import Optional
from fastapi import Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer
from jose import jwt, JWTError
from passlib.context import CryptContext
SECRET_KEY = "xoBZUbHlJ-Tg-IvQw5mHnnELZMNz2iHjQnQ2PNok33g" # Unique key
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 60
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")
# ===== Password Helpers =====
def hash_password(password: str) -> str:
return pwd_context.hash(password)
def verify_password(plain_password: str, hashed_password: str) -> bool:
return pwd_context.verify(plain_password, hashed_password)
# ===== Token Helpers =====
def create_access_token(subject: str, expires_minutes: Optional[int] = None) -> str:
expire = datetime.now(timezone.utc) + timedelta(
minutes=expires_minutes or ACCESS_TOKEN_EXPIRE_MINUTES
)
to_encode = {"sub": subject, "exp": expire}
return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
# ===== Auth Dependency =====
def get_current_user(token: str = Depends(oauth2_scheme)) -> str:
credentials_exception = HTTPException(
status_code=401,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
username: str = payload.get("sub") # type: ignore
if username is None:
raise credentials_exception
except JWTError:
raise credentials_exception
return username
6. API Endpoints (routers/items.py)
Here we define the actual API routes for our "items." We use FastAPI's APIRouter to keep related endpoints organized. Notice the Depends(get_current_user) dependency, which ensures that only authenticated users can access these routes.
# app/routers/items.py
from app.security import get_current_user
from fastapi import APIRouter, Depends
from ..models import ItemCreate, ItemResponse
from .. import crud
router = APIRouter()
@router.post("/", response_model=ItemResponse, status_code=200)
def create_item(payload: ItemCreate, user: str = Depends(get_current_user)):
return crud.create_item(payload)
@router.get("/{item_id}", response_model=ItemResponse)
def get_item(item_id: int, user: str = Depends(get_current_user)):
return crud.read_item(item_id)
# ... other item endpoints: update, delete, get_all
7. Rate Limiting (middleware.py)
To protect our API from abuse, we implement a simple rate-limiting middleware. This middleware restricts each user (identified by IP address) to 5 requests every 60 seconds.
# app/middleware.py
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import JSONResponse
import time
# A simple in-memory store for request timestamps
request_counts = {}
class RateLimitMiddleware(BaseHTTPMiddleware):
def __init__(self, app, limit: int, window_seconds: int):
super().__init__(app)
self.limit = limit
self.window = window_seconds
async def dispatch(self, request: Request, call_next):
client_ip = request.client.host
current_time = time.time()
if client_ip not in request_counts:
request_counts[client_ip] = []
# Filter out old requests
request_counts[client_ip] = [
t for t in request_counts[client_ip] if current_time - t < self.window
]
if len(request_counts[client_ip]) >= self.limit:
return JSONResponse(
status_code=429,
content={"detail": "Rate limit exceeded"}
)
request_counts[client_ip].append(current_time)
response = await call_next(request)
return response
8. The Main App (app/main.py)
Finally, the main.py file ties everything together. It creates the FastAPI instance, registers our exception handlers, adds the rate-limiting middleware, and includes the routers for auth and items.
# app/main.py
from fastapi import FastAPI
from .exceptions import register_exception_handlers
from .middleware import RateLimitMiddleware
from .routers import items, auth
import os
app = FastAPI(title="Minimal FastAPI Application", version="1.0.0")
# Register custom exception handlers
register_exception_handlers(app)
# Add middleware conditionally (not during testing)
if not os.getenv("TESTING"):
app.add_middleware(RateLimitMiddleware, limit=5, window_seconds=60)
# Include the API routers
app.include_router(items.router, prefix="/items", tags=["items"])
app.include_router(auth.router, prefix="/auth", tags=["auth"])
@app.get("/health")
def health():
return {"status": "ok"}
๐ Running and Testing
With the code in place, let's run our API.
Run Locally
Start the development server using Uvicorn:
uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload
The --reload flag automatically restarts the server when you make changes. Now, open your browser and navigate to http://127.0.0.1:8000/docs and http://127.0.0.1:8000/redoc. You'll see the interactive Swagger UI and OpenAPI UI where you can test your endpoints live!
Automated Testing
Run the automated test suite with Pytest. We set the TESTING environment variable to disable rate limiting during tests. Test cases are written inside tests folder.
$env:TESTING="1"
pytest -v
๐ณ Dockerizing the Application
Containerizing our app with Docker ensures it runs consistently across any environment. The Dockerfile defines the steps to build our application image.
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt ./
RUN pip install --no-cache-dir -r requirements.txt
COPY app ./app
EXPOSE 8000
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]
To build and run the Docker container:
# Build the Docker image
docker build -t fastapi_application .
# Run the container
docker run -p 8000:8000 fastapi_application
๐ Conclusion
Congratulations! You now have a robust FastAPI application complete with data validation, authentication, rate limiting, and a deployment-ready Docker image. This structure provides a solid foundation that you can easily extend for more complex projects.
Codebase
GitHub โ https://github.com/Trojan-Dev-AFK/fastapi_application


