FastAPI in Practice: Build a Clean CRUD API with Routers, Dependencies, and Async SQLAlchemy

Pictalingo

·

FastAPI in Practice: Build a Clean CRUD API with Routers, Dependencies, and Async SQLAlchemy

FastAPI is popular because it’s fast to build, strongly typed, and produces interactive API docs automatically. But “hello world” examples don’t show the patterns you’ll need on real projects: structured routers, dependency injection, consistent error handling, and a database layer that won’t become spaghetti.

In this hands-on guide, you’ll build a small but production-shaped REST API for todos using:

  • APIRouter to keep endpoints modular
  • dependencies for DB sessions and authentication
  • async SQLAlchemy for database access
  • Pydantic schemas for clean input/output models

The code below is “working” in the sense that you can copy it into a folder, install dependencies, and run it.

Project layout

Here’s a simple structure that scales:

app/ __init__.py main.py db.py models.py schemas.py deps.py routers/ __init__.py todos.py health.py

Install dependencies

This example uses SQLite with async support for minimal setup. Swap to Postgres later with asyncpg.

python -m venv .venv source .venv/bin/activate # Windows: .venv\Scripts\activate pip install fastapi "uvicorn[standard]" sqlalchemy aiosqlite pydantic

Database setup (async SQLAlchemy)

Create app/db.py. This file defines an async engine and a session factory you’ll inject into endpoints.

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine from sqlalchemy.orm import sessionmaker DATABASE_URL = "sqlite+aiosqlite:///./dev.db" engine = create_async_engine( DATABASE_URL, echo=False, # set True to see SQL while learning future=True, ) AsyncSessionLocal = sessionmaker( bind=engine, class_=AsyncSession, expire_on_commit=False, ) async def get_db() -> AsyncSession: async with AsyncSessionLocal() as session: yield session

Models (SQLAlchemy ORM)

Create app/models.py. We’ll store todos with a title, completion flag, and timestamps.

from sqlalchemy import Boolean, DateTime, Integer, String, func from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column class Base(DeclarativeBase): pass class Todo(Base): __tablename__ = "todos" id: Mapped[int] = mapped_column(Integer, primary_key=True) title: Mapped[str] = mapped_column(String(200), nullable=False) done: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False) created_at: Mapped[str] = mapped_column(DateTime(timezone=True), server_default=func.now()) updated_at: Mapped[str] = mapped_column( DateTime(timezone=True), server_default=func.now(), onupdate=func.now(), )

Schemas (Pydantic models for input/output)

Create app/schemas.py. Separate “what clients send” from “what clients receive”. This helps you avoid accidentally exposing internal fields.

from pydantic import BaseModel, Field class TodoCreate(BaseModel): title: str = Field(min_length=1, max_length=200) class TodoUpdate(BaseModel): title: str | None = Field(default=None, min_length=1, max_length=200) done: bool | None = None class TodoOut(BaseModel): id: int title: str done: bool class Config: from_attributes = True # allows returning ORM objects directly

Dependencies (auth stub + shared helpers)

Create app/deps.py. For junior/mid devs, the key idea is: dependencies are just functions FastAPI can call for you and inject into endpoints. Here we add a tiny API-key check to demonstrate the pattern.

from fastapi import Depends, Header, HTTPException, status API_KEY = "dev-only-change-me" async def require_api_key(x_api_key: str | None = Header(default=None)) -> None: if x_api_key != API_KEY: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid or missing API key", )

In real projects, you might replace this with OAuth2/JWT, but the dependency wiring stays similar.

Routers: health check

Create app/routers/health.py.

from fastapi import APIRouter router = APIRouter(tags=["health"]) @router.get("/health") async def health(): return {"status": "ok"}

Routers: todo CRUD

Create app/routers/todos.py. This shows common patterns:

  • use Depends(get_db) for DB session injection
  • use Depends(require_api_key) for endpoint protection
  • raise HTTPException(404) when a resource doesn’t exist
  • commit + refresh on create/update
from fastapi import APIRouter, Depends, HTTPException, status from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.db import get_db from app.deps import require_api_key from app.models import Todo from app.schemas import TodoCreate, TodoOut, TodoUpdate router = APIRouter(prefix="/todos", tags=["todos"]) @router.get("", response_model=list[TodoOut]) async def list_todos( db: AsyncSession = Depends(get_db), _: None = Depends(require_api_key), ): result = await db.execute(select(Todo).order_by(Todo.id.desc())) return result.scalars().all() @router.post("", response_model=TodoOut, status_code=status.HTTP_201_CREATED) async def create_todo( payload: TodoCreate, db: AsyncSession = Depends(get_db), _: None = Depends(require_api_key), ): todo = Todo(title=payload.title, done=False) db.add(todo) await db.commit() await db.refresh(todo) return todo @router.get("/{todo_id}", response_model=TodoOut) async def get_todo( todo_id: int, db: AsyncSession = Depends(get_db), _: None = Depends(require_api_key), ): result = await db.execute(select(Todo).where(Todo.id == todo_id)) todo = result.scalar_one_or_none() if not todo: raise HTTPException(status_code=404, detail="Todo not found") return todo @router.patch("/{todo_id}", response_model=TodoOut) async def update_todo( todo_id: int, payload: TodoUpdate, db: AsyncSession = Depends(get_db), _: None = Depends(require_api_key), ): result = await db.execute(select(Todo).where(Todo.id == todo_id)) todo = result.scalar_one_or_none() if not todo: raise HTTPException(status_code=404, detail="Todo not found") if payload.title is not None: todo.title = payload.title if payload.done is not None: todo.done = payload.done await db.commit() await db.refresh(todo) return todo @router.delete("/{todo_id}", status_code=status.HTTP_204_NO_CONTENT) async def delete_todo( todo_id: int, db: AsyncSession = Depends(get_db), _: None = Depends(require_api_key), ): result = await db.execute(select(Todo).where(Todo.id == todo_id)) todo = result.scalar_one_or_none() if not todo: raise HTTPException(status_code=404, detail="Todo not found") await db.delete(todo) await db.commit() return None

App entrypoint (main.py) + create tables on startup

Create app/main.py. For a tutorial, we’ll create tables on startup. In production you’d likely use migrations, but this keeps the example runnable.

from fastapi import FastAPI from app.db import engine from app.models import Base from app.routers.health import router as health_router from app.routers.todos import router as todos_router app = FastAPI(title="Todo API", version="1.0") @app.on_event("startup") async def on_startup(): async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) app.include_router(health_router) app.include_router(todos_router)

Run it

uvicorn app.main:app --reload

Open the docs at http://127.0.0.1:8000/docs.

Try it with curl

Remember: endpoints require the header X-API-Key matching dev-only-change-me.

# Health (no auth) curl http://127.0.0.1:8000/health # Create a todo curl -X POST http://127.0.0.1:8000/todos \ -H "Content-Type: application/json" \ -H "X-API-Key: dev-only-change-me" \ -d '{"title":"Ship the feature"}' # List todos curl http://127.0.0.1:8000/todos \ -H "X-API-Key: dev-only-change-me" # Update a todo curl -X PATCH http://127.0.0.1:8000/todos/1 \ -H "Content-Type: application/json" \ -H "X-API-Key: dev-only-change-me" \ -d '{"done": true}'

Patterns to keep your FastAPI codebase healthy

  • Keep routers small and focused. If a router file grows past ~200–300 lines, split it (e.g., todos_read.py, todos_write.py) or introduce a service layer.

  • Use schemas as a contract. Returning TodoOut avoids leaking internal columns and makes changes safer.

  • Dependencies are your “glue”. DB sessions, auth, feature flags, tenant resolution, request IDs—these belong in dependencies instead of copied logic in every endpoint.

  • Be explicit about 404 vs 400. Missing resource? 404. Bad input? 422 (FastAPI does this) or 400 if you validate manually.

  • Prefer async all the way. If you start async (FastAPI + async SQLAlchemy), keep I/O async (HTTP calls, DB calls) or you’ll lose the benefits.

Next upgrades (when you’re ready)

If you want to take this from “tutorial” to “real service,” add these in order:

  • Migrations with Alembic so schema changes are versioned

  • Config management (env vars) for DB URL and API keys

  • Logging + request IDs for traceability

  • Pagination on list endpoints (limit/offset or cursor-based)

  • Testing with pytest + httpx and a temporary database

At this point you’ve got the core mechanics that matter: clean routing, dependency injection, typed schemas, and a database layer you can extend without rewriting everything.

Leave a Reply

Your email address will not be published. Required fields are marked *