FastAPI CRUD API with SQLAlchemy 2.0 (Async), Pydantic v2, and Real Tests

FastAPI CRUD API with SQLAlchemy 2.0 (Async), Pydantic v2, and Real Tests

If you’re building a web API, “it runs locally” isn’t the finish line. You want predictable validation, clean database access, and tests that don’t touch production resources. In this hands-on guide, you’ll build a small but production-shaped FastAPI service using:

• FastAPI for routing + dependency injection
• SQLAlchemy 2.0 async ORM for database work
• Alembic-friendly project structure (we’ll structure code for it)
• Pydantic v2 schemas for input/output
• Proper HTTP errors and pagination
• Tests with dependency overrides (no real DB required in unit tests)

We’ll build a simple “Tasks” API: create, list (paginated), update, delete.

1) Project setup

Install dependencies:

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

We’ll use SQLite for simplicity (aiosqlite), but the patterns translate to Postgres with asyncpg.

Suggested layout:

app/ main.py db.py models.py schemas.py crud.py routes.py tests/ test_tasks.py

2) Database layer (async session dependency)

Create app/db.py. This sets up the SQLAlchemy async engine and a FastAPI dependency that yields a session per request.

# app/db.py from typing import AsyncGenerator from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine DATABASE_URL = "sqlite+aiosqlite:///./dev.db" engine = create_async_engine( DATABASE_URL, echo=False, # flip to True when debugging SQL future=True, ) AsyncSessionLocal = async_sessionmaker( engine, expire_on_commit=False, class_=AsyncSession, ) async def get_session() -> AsyncGenerator[AsyncSession, None]: async with AsyncSessionLocal() as session: yield session

expire_on_commit=False prevents ORM objects from becoming unusable after commit (a common beginner “why is this None?” pain).

3) ORM model

Create app/models.py. SQLAlchemy 2.0 style uses Mapped and mapped_column.

# app/models.py from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column from sqlalchemy import String, Boolean, Integer class Base(DeclarativeBase): pass class Task(Base): __tablename__ = "tasks" 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)

In a real project, you’d run migrations via Alembic. For this tutorial, we’ll auto-create tables at startup (fine for demos; swap to migrations later).

4) Pydantic v2 schemas

Create app/schemas.py. Use separate schemas for create/update/output to avoid over-posting and accidental field exposure.

# app/schemas.py from pydantic import BaseModel, Field, ConfigDict class TaskCreate(BaseModel): title: str = Field(min_length=1, max_length=200) class TaskUpdate(BaseModel): title: str | None = Field(default=None, min_length=1, max_length=200) done: bool | None = None class TaskOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int title: str done: bool

from_attributes=True enables returning ORM objects directly as response models.

5) CRUD functions (keep routes thin)

Create app/crud.py. Keeping DB logic out of routes makes code easier to test and refactor.

# app/crud.py from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select, func from app.models import Task from app.schemas import TaskCreate, TaskUpdate async def create_task(session: AsyncSession, data: TaskCreate) -> Task: task = Task(title=data.title) session.add(task) await session.commit() await session.refresh(task) return task async def get_task(session: AsyncSession, task_id: int) -> Task | None: result = await session.execute(select(Task).where(Task.id == task_id)) return result.scalar_one_or_none() async def update_task(session: AsyncSession, task: Task, data: TaskUpdate) -> Task: if data.title is not None: task.title = data.title if data.done is not None: task.done = data.done await session.commit() await session.refresh(task) return task async def delete_task(session: AsyncSession, task: Task) -> None: await session.delete(task) await session.commit() async def list_tasks( session: AsyncSession, limit: int, offset: int, ) -> tuple[list[Task], int]: # total count total_result = await session.execute(select(func.count()).select_from(Task)) total = int(total_result.scalar_one()) # page data rows_result = await session.execute( select(Task).order_by(Task.id).limit(limit).offset(offset) ) tasks = list(rows_result.scalars().all()) return tasks, total

6) Routes with validation, pagination, and correct HTTP errors

Create app/routes.py. Note the use of query bounds (ge, le) to protect your API from someone requesting limit=999999.

# app/routes.py from fastapi import APIRouter, Depends, HTTPException, Query, status from sqlalchemy.ext.asyncio import AsyncSession from app.db import get_session from app.schemas import TaskCreate, TaskUpdate, TaskOut from app import crud router = APIRouter(prefix="/tasks", tags=["tasks"]) @router.post("", response_model=TaskOut, status_code=status.HTTP_201_CREATED) async def create_task( payload: TaskCreate, session: AsyncSession = Depends(get_session), ): return await crud.create_task(session, payload) @router.get("", response_model=dict) async def list_tasks( limit: int = Query(20, ge=1, le=100), offset: int = Query(0, ge=0), session: AsyncSession = Depends(get_session), ): tasks, total = await crud.list_tasks(session, limit=limit, offset=offset) return { "items": [TaskOut.model_validate(t) for t in tasks], "total": total, "limit": limit, "offset": offset, } @router.patch("/{task_id}", response_model=TaskOut) async def patch_task( task_id: int, payload: TaskUpdate, session: AsyncSession = Depends(get_session), ): task = await crud.get_task(session, task_id) if not task: raise HTTPException(status_code=404, detail="Task not found") return await crud.update_task(session, task, payload) @router.delete("/{task_id}", status_code=status.HTTP_204_NO_CONTENT) async def delete_task( task_id: int, session: AsyncSession = Depends(get_session), ): task = await crud.get_task(session, task_id) if not task: raise HTTPException(status_code=404, detail="Task not found") await crud.delete_task(session, task) return None

Notice we return a small pagination envelope. This makes frontends simpler because they don’t have to infer totals.

7) App entrypoint + table creation for local dev

Create app/main.py. For local dev we’ll create tables on startup. Later, replace this with Alembic migrations.

# app/main.py from fastapi import FastAPI from app.routes import router from app.db import engine from app.models import Base app = FastAPI(title="Tasks API") app.include_router(router) @app.on_event("startup") async def on_startup() -> None: async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all)

Run it:

uvicorn app.main:app --reload

Open the interactive docs at /docs and try creating/listing tasks.

8) Tests that don’t touch your dev DB (dependency override)

The most practical FastAPI testing trick: override dependencies. Here we’ll create a separate in-memory SQLite database for tests and override get_session.

Create tests/test_tasks.py:

# tests/test_tasks.py import pytest from httpx import AsyncClient from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession from app.main import app from app.db import get_session from app.models import Base TEST_DB_URL = "sqlite+aiosqlite:///:memory:" @pytest.fixture async def test_session() -> AsyncSession: engine = create_async_engine(TEST_DB_URL, future=True) async_session = async_sessionmaker(engine, expire_on_commit=False) async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) async with async_session() as session: yield session await engine.dispose() @pytest.fixture def override_get_session(test_session: AsyncSession): async def _override(): yield test_session app.dependency_overrides[get_session] = _override yield app.dependency_overrides.clear() @pytest.mark.asyncio async def test_create_and_list_tasks(override_get_session): async with AsyncClient(app=app, base_url="http://test") as client: # create r = await client.post("/tasks", json={"title": "Write tests"}) assert r.status_code == 201 data = r.json() assert data["title"] == "Write tests" assert data["done"] is False # list r = await client.get("/tasks?limit=10&offset=0") assert r.status_code == 200 payload = r.json() assert payload["total"] == 1 assert len(payload["items"]) == 1 assert payload["items"][0]["title"] == "Write tests" @pytest.mark.asyncio async def test_patch_and_delete(override_get_session): async with AsyncClient(app=app, base_url="http://test") as client: r = await client.post("/tasks", json={"title": "Ship it"}) task_id = r.json()["id"] r = await client.patch(f"/tasks/{task_id}", json={"done": True}) assert r.status_code == 200 assert r.json()["done"] is True r = await client.delete(f"/tasks/{task_id}") assert r.status_code == 204 r = await client.patch(f"/tasks/{task_id}", json={"done": False}) assert r.status_code == 404

Run tests:

pytest -q

This pattern scales well: you can swap the DB URL to Postgres in CI, or use transaction rollbacks per test for speed.

9) Practical finishing touches (what juniors forget)

• Input constraints: use Query(..., le=100) and Field(max_length=...) to prevent abuse and accidental huge payloads.

• Consistent errors: always 404 missing resources; avoid returning 200 with “not found” messages.

• Keep handlers thin: move DB logic to crud.py so you can unit test without HTTP.

• Don’t skip pagination envelopes: returning {items, total, limit, offset} saves frontend time and avoids “infinite scroll lies”.

• Plan for migrations: table creation on startup is fine for demos, but for teams use Alembic so schema changes are tracked.

Wrap-up

You now have a small, reliable FastAPI service with async SQLAlchemy, clean schemas, paginated listing, proper error handling, and tests that run against a separate database via dependency overrides. From here, common next steps are adding authentication (JWT or session), switching SQLite to Postgres, and introducing Alembic migrations.