FastAPI CRUD You Can Actually Ship: Async SQLAlchemy, Migrations, Pagination, and Tests

FastAPI CRUD You Can Actually Ship: Async SQLAlchemy, Migrations, Pagination, and Tests

FastAPI is great for building APIs quickly, but junior/mid devs often get stuck on the “okay… now what?” part: database setup, clean project structure, pagination, and tests that don’t turn into a mess. This hands-on guide walks you through a small but realistic CRUD API using FastAPI + async SQLAlchemy, with pagination and integration tests.

We’ll build a tiny “notes” service:

• Create, list (paginated), fetch, update, delete notes
• Async DB access (SQLite for local dev, easily swap to Postgres)
• Clean structure with routers and dependencies
• Pytest integration tests using FastAPI’s test client

1) Install dependencies

Create a virtual environment, then install:

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

Why these?

• sqlalchemy for ORM (async engine)
• aiosqlite for async SQLite driver
• httpx for async-friendly testing

2) Project layout

Keep it simple but scalable:

app/ __init__.py main.py db.py models.py schemas.py deps.py routers/ __init__.py notes.py tests/ test_notes.py

3) Database setup (async SQLAlchemy)

Create app/db.py:

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine from sqlalchemy.orm import sessionmaker, DeclarativeBase DATABASE_URL = "sqlite+aiosqlite:///./dev.db" engine = create_async_engine( DATABASE_URL, echo=False, # set True when debugging SQL ) async_session_factory = sessionmaker( engine, expire_on_commit=False, class_=AsyncSession, ) class Base(DeclarativeBase): pass

Then create app/models.py:

from sqlalchemy import String, Text from sqlalchemy.orm import Mapped, mapped_column from .db import Base class Note(Base): __tablename__ = "notes" id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True) title: Mapped[str] = mapped_column(String(200)) body: Mapped[str] = mapped_column(Text(), default="")

In real apps, you’ll likely use Alembic for migrations. For this tutorial, we’ll auto-create tables on startup (fine for local/dev). You can swap to Alembic later without changing your API code.

4) Pydantic schemas for input/output

Create app/schemas.py:

from pydantic import BaseModel, Field class NoteCreate(BaseModel): title: str = Field(min_length=1, max_length=200) body: str = Field(default="", max_length=50_000) class NoteUpdate(BaseModel): title: str | None = Field(default=None, min_length=1, max_length=200) body: str | None = Field(default=None, max_length=50_000) class NoteOut(BaseModel): id: int title: str body: str class Config: from_attributes = True class Page(BaseModel): items: list[NoteOut] total: int limit: int offset: int

Tip: from_attributes = True lets Pydantic convert ORM objects cleanly.

5) Dependency injection for DB sessions

Create app/deps.py:

from collections.abc import AsyncGenerator from sqlalchemy.ext.asyncio import AsyncSession from .db import async_session_factory async def get_db() -> AsyncGenerator[AsyncSession, None]: async with async_session_factory() as session: yield session

This is a big FastAPI win: your endpoints stay clean and testable.

6) Notes router (CRUD + pagination)

Create app/routers/notes.py:

from fastapi import APIRouter, Depends, HTTPException, Query, status from sqlalchemy import select, func from sqlalchemy.ext.asyncio import AsyncSession from ..deps import get_db from ..models import Note from ..schemas import NoteCreate, NoteUpdate, NoteOut, Page router = APIRouter(prefix="/notes", tags=["notes"]) @router.post("", response_model=NoteOut, status_code=status.HTTP_201_CREATED) async def create_note(payload: NoteCreate, db: AsyncSession = Depends(get_db)): note = Note(title=payload.title, body=payload.body) db.add(note) await db.commit() await db.refresh(note) return note @router.get("", response_model=Page) async def list_notes( limit: int = Query(default=20, ge=1, le=100), offset: int = Query(default=0, ge=0), db: AsyncSession = Depends(get_db), ): total_stmt = select(func.count()).select_from(Note) total = (await db.execute(total_stmt)).scalar_one() stmt = select(Note).order_by(Note.id.desc()).limit(limit).offset(offset) items = (await db.execute(stmt)).scalars().all() return {"items": items, "total": total, "limit": limit, "offset": offset} @router.get("/{note_id}", response_model=NoteOut) async def get_note(note_id: int, db: AsyncSession = Depends(get_db)): note = await db.get(Note, note_id) if not note: raise HTTPException(status_code=404, detail="Note not found") return note @router.patch("/{note_id}", response_model=NoteOut) async def update_note(note_id: int, payload: NoteUpdate, db: AsyncSession = Depends(get_db)): note = await db.get(Note, note_id) if not note: raise HTTPException(status_code=404, detail="Note not found") if payload.title is not None: note.title = payload.title if payload.body is not None: note.body = payload.body await db.commit() await db.refresh(note) return note @router.delete("/{note_id}", status_code=status.HTTP_204_NO_CONTENT) async def delete_note(note_id: int, db: AsyncSession = Depends(get_db)): note = await db.get(Note, note_id) if not note: raise HTTPException(status_code=404, detail="Note not found") await db.delete(note) await db.commit() return None

Pagination here is “offset/limit”—easy for beginners and fine for many apps. When you hit performance limits (large offsets), you can migrate to cursor-based pagination without changing your route shape much.

7) Wire it up in the FastAPI app

Create app/main.py:

from fastapi import FastAPI from .db import engine, Base from .routers.notes import router as notes_router app = FastAPI(title="Notes API") @app.on_event("startup") async def on_startup(): # Dev-friendly: create tables automatically. async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) app.include_router(notes_router)

Run it:

uvicorn app.main:app --reload

Try:

• POST /notes with JSON {"title":"Hello","body":"World"}
• GET /notes?limit=10&offset=0
• PATCH /notes/1 with {"body":"Updated"}

8) Integration tests with pytest

Testing FastAPI works best when you test your actual HTTP API. We’ll use an in-memory SQLite database for speed and isolation.

Create tests/test_notes.py:

import pytest from httpx import AsyncClient from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession from sqlalchemy.orm import sessionmaker from app.main import app from app.db import Base from app.deps import get_db TEST_DB_URL = "sqlite+aiosqlite:///:memory:" @pytest.fixture async def test_client(): engine = create_async_engine(TEST_DB_URL, echo=False) async_session_factory = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) async def override_get_db(): async with async_session_factory() as session: yield session app.dependency_overrides[get_db] = override_get_db async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) async with AsyncClient(app=app, base_url="http://test") as client: yield client app.dependency_overrides.clear() await engine.dispose() @pytest.mark.asyncio async def test_create_and_list_notes(test_client): # Create res = await test_client.post("/notes", json={"title": "A", "body": "B"}) assert res.status_code == 201 note = res.json() assert note["id"] == 1 assert note["title"] == "A" # List res = await test_client.get("/notes?limit=20&offset=0") assert res.status_code == 200 page = res.json() assert page["total"] == 1 assert len(page["items"]) == 1 assert page["items"][0]["title"] == "A" @pytest.mark.asyncio async def test_update_and_delete(test_client): res = await test_client.post("/notes", json={"title": "T", "body": "X"}) note_id = res.json()["id"] res = await test_client.patch(f"/notes/{note_id}", json={"body": "Y"}) assert res.status_code == 200 assert res.json()["body"] == "Y" res = await test_client.delete(f"/notes/{note_id}") assert res.status_code == 204 res = await test_client.get(f"/notes/{note_id}") assert res.status_code == 404

Run tests:

pytest -q

9) Practical “next steps” for production

This starter kit is intentionally small, but it’s ready to grow. If you’re moving beyond local/dev, prioritize these upgrades:

• Real migrations: add Alembic so schema changes are explicit and reversible.

• Config via environment: move DATABASE_URL into settings (pydantic-settings) and load from .env.

• Validation and errors: keep using Pydantic constraints; return consistent error shapes for clients.

• Auth: add JWT or session auth using dependencies (FastAPI makes this clean).

• Logging/observability: structured logs, request IDs, and metrics (Prometheus/OpenTelemetry) when you’re ready.

Wrap-up

You now have a practical FastAPI pattern: async DB sessions via dependencies, clean routers, paginated list endpoints, and integration tests that validate your real HTTP behavior. The best part is that none of this is “toy code”—it’s the same shape you’ll use for bigger services, just with more endpoints and more modules.