FastAPI in Practice: Build a Production-Ready REST API with Dependencies, Pagination, and Tests

FastAPI in Practice: Build a Production-Ready REST API with Dependencies, Pagination, and Tests

FastAPI is popular because it’s fast to build with, easy to validate inputs, and naturally fits modern Python patterns (type hints, async, dependency injection). In this hands-on guide, you’ll build a small but “real” API: a todos service with validation, paging, simple token auth, clean error handling, and tests. The examples are intentionally practical for junior/mid developers.

What you’ll implement:

• CRUD endpoints for todos
• Pydantic models for request/response validation
• Dependency injection (DB/session + auth)
• Offset pagination + filtering
• Centralized error handling
• Pytest + HTTPX tests

Project setup

Create a minimal structure:

app/ __init__.py main.py deps.py models.py repo.py errors.py tests/ test_todos.py pyproject.toml (or requirements.txt)

Install dependencies:

pip install fastapi uvicorn pydantic sqlalchemy aiosqlite httpx pytest pytest-asyncio

We’ll use async SQLAlchemy with SQLite so the code stays self-contained and still resembles production patterns.

Define data models (validation done right)

FastAPI leans on Pydantic models for input/output. This is where you enforce shape, types, and constraints.

# app/models.py from datetime import datetime from typing import Optional from pydantic import BaseModel, Field class TodoCreate(BaseModel): title: str = Field(min_length=1, max_length=200) details: Optional[str] = Field(default=None, max_length=2000) class TodoUpdate(BaseModel): title: Optional[str] = Field(default=None, min_length=1, max_length=200) details: Optional[str] = Field(default=None, max_length=2000) done: Optional[bool] = None class TodoOut(BaseModel): id: int title: str details: Optional[str] done: bool created_at: datetime class PageOut(BaseModel): items: list[TodoOut] total: int offset: int limit: int

Tip: keep “create/update” models separate. It prevents accidental writes of server-managed fields like id or created_at.

Database layer (async SQLAlchemy + repository)

Separate DB logic from route handlers. Your endpoints stay readable, and testing becomes easier.

# app/repo.py from datetime import datetime from typing import Optional, Tuple, List from sqlalchemy import ( Boolean, Column, DateTime, Integer, String, Text, select, func, update, delete ) from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import declarative_base Base = declarative_base() class Todo(Base): __tablename__ = "todos" id = Column(Integer, primary_key=True) title = Column(String(200), nullable=False) details = Column(Text, nullable=True) done = Column(Boolean, nullable=False, default=False) created_at = Column(DateTime, nullable=False, default=datetime.utcnow) async def create_todo(session: AsyncSession, title: str, details: Optional[str]): todo = Todo(title=title, details=details, done=False) session.add(todo) await session.commit() await session.refresh(todo) return todo async def get_todo(session: AsyncSession, todo_id: int): res = await session.execute(select(Todo).where(Todo.id == todo_id)) return res.scalar_one_or_none() async def list_todos( session: AsyncSession, offset: int, limit: int, done: Optional[bool] = None, q: Optional[str] = None, ) -> Tuple[List[Todo], int]: stmt = select(Todo) count_stmt = select(func.count()).select_from(Todo) if done is not None: stmt = stmt.where(Todo.done == done) count_stmt = count_stmt.where(Todo.done == done) if q: like = f"%{q}%" stmt = stmt.where(Todo.title.like(like)) count_stmt = count_stmt.where(Todo.title.like(like)) total = (await session.execute(count_stmt)).scalar_one() stmt = stmt.order_by(Todo.id.desc()).offset(offset).limit(limit) items = (await session.execute(stmt)).scalars().all() return items, total async def update_todo(session: AsyncSession, todo_id: int, **fields): fields = {k: v for k, v in fields.items() if v is not None} if not fields: return await get_todo(session, todo_id) await session.execute(update(Todo).where(Todo.id == todo_id).values(**fields)) await session.commit() return await get_todo(session, todo_id) async def delete_todo(session: AsyncSession, todo_id: int) -> bool: res = await session.execute(delete(Todo).where(Todo.id == todo_id)) await session.commit() return res.rowcount > 0

Dependencies: database session + simple auth

FastAPI’s dependency injection is one of its biggest strengths. You can share cross-cutting concerns (DB sessions, auth, rate limits, feature flags) without copying code into every route.

# app/deps.py from typing import AsyncGenerator, Optional from fastapi import Depends, Header, HTTPException, status from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession from .repo import Base DATABASE_URL = "sqlite+aiosqlite:///./app.db" engine = create_async_engine(DATABASE_URL, echo=False) SessionLocal = async_sessionmaker(engine, expire_on_commit=False) async def init_db() -> None: async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) async def get_session() -> AsyncGenerator[AsyncSession, None]: async with SessionLocal() as session: yield session async def require_token(x_api_key: Optional[str] = Header(default=None)) -> str: # In production, validate JWT or lookup an API key in DB/Redis. if x_api_key != "dev-secret": raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid or missing API key", ) return x_api_key

Now any route that needs auth can simply add Depends(require_token).

Centralized errors (clean 404s and consistent responses)

It’s common to sprinkle HTTPException(404) everywhere. A small helper keeps it consistent.

# app/errors.py from fastapi import HTTPException, status def not_found(resource: str = "Item") -> HTTPException: return HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"{resource} not found", )

Build the API routes

With the repo and dependencies in place, the routes stay straightforward.

# app/main.py from typing import Optional from fastapi import FastAPI, Depends, Query from sqlalchemy.ext.asyncio import AsyncSession from .deps import get_session, init_db, require_token from .models import TodoCreate, TodoUpdate, TodoOut, PageOut from . import repo from .errors import not_found app = FastAPI(title="Todos API", version="1.0.0") @app.on_event("startup") async def startup(): await init_db() def to_out(todo: repo.Todo) -> TodoOut: return TodoOut( id=todo.id, title=todo.title, details=todo.details, done=todo.done, created_at=todo.created_at, ) @app.post("/todos", response_model=TodoOut, dependencies=[Depends(require_token)]) async def create_todo( payload: TodoCreate, session: AsyncSession = Depends(get_session), ): todo = await repo.create_todo(session, title=payload.title, details=payload.details) return to_out(todo) @app.get("/todos/{todo_id}", response_model=TodoOut, dependencies=[Depends(require_token)]) async def get_todo(todo_id: int, session: AsyncSession = Depends(get_session)): todo = await repo.get_todo(session, todo_id) if not todo: raise not_found("Todo") return to_out(todo) @app.get("/todos", response_model=PageOut, dependencies=[Depends(require_token)]) async def list_todos( session: AsyncSession = Depends(get_session), offset: int = Query(0, ge=0), limit: int = Query(20, ge=1, le=100), done: Optional[bool] = None, q: Optional[str] = Query(default=None, max_length=200), ): items, total = await repo.list_todos(session, offset=offset, limit=limit, done=done, q=q) return PageOut( items=[to_out(t) for t in items], total=total, offset=offset, limit=limit, ) @app.patch("/todos/{todo_id}", response_model=TodoOut, dependencies=[Depends(require_token)]) async def patch_todo( todo_id: int, payload: TodoUpdate, session: AsyncSession = Depends(get_session), ): updated = await repo.update_todo(session, todo_id, **payload.model_dump()) if not updated: raise not_found("Todo") return to_out(updated) @app.delete("/todos/{todo_id}", response_model=dict, dependencies=[Depends(require_token)]) async def remove_todo(todo_id: int, session: AsyncSession = Depends(get_session)): ok = await repo.delete_todo(session, todo_id) if not ok: raise not_found("Todo") return {"deleted": True}

Run it:

uvicorn app.main:app --reload

Try requests (note the header X-API-Key):

curl -X POST "http://127.0.0.1:8000/todos" \ -H "Content-Type: application/json" \ -H "X-API-Key: dev-secret" \ -d '{"title":"Ship FastAPI tutorial","details":"Write tests too"}'

Testing with pytest + HTTPX (async)

FastAPI apps are easiest to test with HTTPX. You can override dependencies to use a test DB and avoid real secrets.

# tests/test_todos.py import pytest from httpx import AsyncClient from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker from app.main import app from app.deps import get_session, require_token from app.repo import Base TEST_DB_URL = "sqlite+aiosqlite:///./test.db" engine = create_async_engine(TEST_DB_URL, echo=False) SessionLocal = async_sessionmaker(engine, expire_on_commit=False) async def override_get_session(): async with SessionLocal() as session: yield session async def override_require_token(): return "test-secret" @pytest.fixture(scope="session", autouse=True) async def setup_db(): async with engine.begin() as conn: await conn.run_sync(Base.metadata.drop_all) await conn.run_sync(Base.metadata.create_all) app.dependency_overrides[get_session] = override_get_session app.dependency_overrides[require_token] = override_require_token yield app.dependency_overrides.clear() @pytest.mark.asyncio async def test_create_and_list(): async with AsyncClient(app=app, base_url="http://test") as client: r = await client.post("/todos", json={"title": "Test todo"}) assert r.status_code == 200 todo = r.json() assert todo["title"] == "Test todo" assert todo["done"] is False r2 = await client.get("/todos?offset=0&limit=10") assert r2.status_code == 200 page = r2.json() assert page["total"] >= 1 assert len(page["items"]) >= 1 @pytest.mark.asyncio async def test_patch_and_delete(): async with AsyncClient(app=app, base_url="http://test") as client: created = (await client.post("/todos", json={"title": "Patch me"})).json() tid = created["id"] r = await client.patch(f"/todos/{tid}", json={"done": True}) assert r.status_code == 200 assert r.json()["done"] is True d = await client.delete(f"/todos/{tid}") assert d.status_code == 200 assert d.json()["deleted"] is True

Run tests:

pytest -q

Practical tips you’ll use on real teams

• Keep routes thin: push DB logic into a repository/service layer to reduce duplicate code and simplify tests.
• Use dependency overrides in tests: they let you swap DB connections and auth without changing production code.
• Validate at the edge: let Pydantic reject bad input early; it reduces defensive code inside your business logic.
• Set pagination limits: using limit constraints (like le=100) prevents accidental “return everything” endpoints.
• Plan auth growth: today it’s a header check; tomorrow it might be JWT validation or an API key lookup. The dependency pattern scales.

Where to go next

If you want to level this up for production, your next incremental steps are:

• Add structured logging and request IDs
• Introduce migrations (Alembic) instead of create_all on startup
• Replace the demo auth with JWT or API keys stored in a database/secret manager
• Add rate limiting (as a dependency/middleware)
• Use a real DB (PostgreSQL) and proper connection pooling

You now have a working FastAPI template that’s small, testable, and built around patterns teams actually use.