Skip to main content

Command Palette

Search for a command to run...

Building a Robust FastAPI Application: From CRUD to Docker

Updated
โ€ข7 min readโ€ขView as Markdown
Building a Robust FastAPI Application: From CRUD to Docker
A
Senior Associate at BMW Techworks India with 6+ years of experience engineering scalable Python backends and containerized automation systems. Passionate about performance optimization, infrastructure security, and integrating LLM/RAG solutions to solve real-world technical bottlenecks. When I'm not auditing GenAI applications or streamlining DevOps pipelines, I write technical blogs to share insights with the developer community.

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.

  1. 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.ps1
    
  2. Install the required packages from the requirements.txt file.

     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