FastAPI Development: Build a Clean CRUD API with Async SQLAlchemy (2.0), Pydantic, and Pagination

FastAPI Development: Build a Clean CRUD API with Async SQLAlchemy (2.0), Pydantic, and Pagination

If you’ve built a couple of FastAPI endpoints, the next “real project” jump is structuring your app so it stays readable as it grows: clean dependencies, a database session per request, consistent validation, and practical patterns like pagination and partial updates.

This hands-on guide walks through a small but production-shaped API for “projects” using:

• FastAPI for routing and dependency injection
• SQLAlchemy 2.0 async ORM for database access
• Pydantic models for request/response validation
• Offset/limit pagination and safe partial updates

We’ll use PostgreSQL in examples, but the code works similarly with SQLite (async driver) for local dev.

1) Install dependencies

Create a virtual environment, then install:

pip install fastapi uvicorn[standard] sqlalchemy[asyncio] asyncpg pydantic python-dotenv

If you want migrations later, also install Alembic:

pip install alembic

2) Project layout (simple and scalable)

A small structure that won’t collapse at 20+ endpoints:

app/ __init__.py main.py db.py models.py schemas.py routers/ __init__.py projects.py .env

Keep it boring: separate DB setup, models, schemas, and routers.

3) Database setup: async engine + session dependency

Create app/db.py:

from __future__ import annotations import os from typing import AsyncGenerator from dotenv import load_dotenv from sqlalchemy.ext.asyncio import ( AsyncSession, async_sessionmaker, create_async_engine, ) from sqlalchemy.orm import DeclarativeBase load_dotenv() DATABASE_URL = os.getenv("DATABASE_URL", "postgresql+asyncpg://user:pass@localhost:5432/mydb") engine = create_async_engine( DATABASE_URL, echo=False, # set True to debug SQL in dev pool_pre_ping=True, # helps avoid stale connections ) SessionLocal = async_sessionmaker( bind=engine, class_=AsyncSession, expire_on_commit=False, ) class Base(DeclarativeBase): pass async def get_db() -> AsyncGenerator[AsyncSession, None]: async with SessionLocal() as session: yield session

Why this matters:

• One AsyncSession per request via dependency injection.
• expire_on_commit=False prevents objects from becoming unusable after commit.
• pool_pre_ping reduces random “connection closed” issues.

4) Define the ORM model

Create app/models.py:

from __future__ import annotations from datetime import datetime from sqlalchemy import String, Text, DateTime, func from sqlalchemy.orm import Mapped, mapped_column from .db import Base class Project(Base): __tablename__ = "projects" id: Mapped[int] = mapped_column(primary_key=True) name: Mapped[str] = mapped_column(String(120), unique=True, index=True) description: Mapped[str | None] = mapped_column(Text(), nullable=True) created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) updated_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), server_default=func.now(), onupdate=func.now(), )

Notes:

• unique=True on name prevents duplicates (we’ll handle the error gracefully).
• index=True helps lookups and search filters.

5) Pydantic schemas for input/output

Create app/schemas.py:

from __future__ import annotations from datetime import datetime from pydantic import BaseModel, Field class ProjectCreate(BaseModel): name: str = Field(min_length=2, max_length=120) description: str | None = Field(default=None, max_length=5000) class ProjectUpdate(BaseModel): # Partial update: all fields optional name: str | None = Field(default=None, min_length=2, max_length=120) description: str | None = Field(default=None, max_length=5000) class ProjectOut(BaseModel): id: int name: str description: str | None created_at: datetime updated_at: datetime | None = None # Pydantic can read ORM objects when this is enabled model_config = {"from_attributes": True} class PaginatedProjects(BaseModel): items: list[ProjectOut] total: int limit: int offset: int

Two key patterns:

• ProjectUpdate supports PATCH-style updates without forcing all fields.
• PaginatedProjects standardizes pagination responses across endpoints.

6) CRUD router with pagination and safe partial updates

Create app/routers/projects.py:

from __future__ import annotations from fastapi import APIRouter, Depends, HTTPException, Query, status from sqlalchemy import select, func from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession from ..db import get_db from ..models import Project from ..schemas import ProjectCreate, ProjectOut, ProjectUpdate, PaginatedProjects router = APIRouter(prefix="/projects", tags=["projects"]) @router.post("", response_model=ProjectOut, status_code=status.HTTP_201_CREATED) async def create_project(payload: ProjectCreate, db: AsyncSession = Depends(get_db)): project = Project(name=payload.name, description=payload.description) db.add(project) try: await db.commit() except IntegrityError: await db.rollback() raise HTTPException(status_code=409, detail="Project name already exists") await db.refresh(project) return project @router.get("", response_model=PaginatedProjects) async def list_projects( db: AsyncSession = Depends(get_db), q: str | None = Query(default=None, description="Search by name (substring)"), limit: int = Query(default=20, ge=1, le=100), offset: int = Query(default=0, ge=0), ): base_stmt = select(Project) if q: base_stmt = base_stmt.where(Project.name.ilike(f"%{q}%")) total_stmt = select(func.count()).select_from(base_stmt.subquery()) total = (await db.execute(total_stmt)).scalar_one() items_stmt = base_stmt.order_by(Project.id.desc()).limit(limit).offset(offset) items = (await db.execute(items_stmt)).scalars().all() return {"items": items, "total": total, "limit": limit, "offset": offset} @router.get("/{project_id}", response_model=ProjectOut) async def get_project(project_id: int, db: AsyncSession = Depends(get_db)): stmt = select(Project).where(Project.id == project_id) project = (await db.execute(stmt)).scalar_one_or_none() if not project: raise HTTPException(status_code=404, detail="Project not found") return project @router.patch("/{project_id}", response_model=ProjectOut) async def update_project( project_id: int, payload: ProjectUpdate, db: AsyncSession = Depends(get_db), ): stmt = select(Project).where(Project.id == project_id) project = (await db.execute(stmt)).scalar_one_or_none() if not project: raise HTTPException(status_code=404, detail="Project not found") # Only update fields that were actually provided data = payload.model_dump(exclude_unset=True) for key, value in data.items(): setattr(project, key, value) try: await db.commit() except IntegrityError: await db.rollback() raise HTTPException(status_code=409, detail="Project name already exists") await db.refresh(project) return project @router.delete("/{project_id}", status_code=status.HTTP_204_NO_CONTENT) async def delete_project(project_id: int, db: AsyncSession = Depends(get_db)): stmt = select(Project).where(Project.id == project_id) project = (await db.execute(stmt)).scalar_one_or_none() if not project: raise HTTPException(status_code=404, detail="Project not found") await db.delete(project) await db.commit() return None

What junior/mid devs usually miss (and this fixes):

• IntegrityError handling for unique constraints (return 409 instead of 500).
• Pagination includes total so UIs can render page numbers properly.
• exclude_unset=True prevents overwriting fields with null when doing partial updates.

7) Wire everything in main.py + create tables

Create app/main.py:

from __future__ import annotations from fastapi import FastAPI from .db import engine, Base from .routers.projects import router as projects_router app = FastAPI(title="Projects API") app.include_router(projects_router) @app.on_event("startup") async def on_startup(): # For real apps, prefer Alembic migrations. # This is fine for a small demo or internal tool. async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all)

Set your .env:

DATABASE_URL=postgresql+asyncpg://user:pass@localhost:5432/mydb

Run the server:

uvicorn app.main:app --reload

8) Try it with curl (realistic workflow)

Create a project:

curl -X POST "http://127.0.0.1:8000/projects" \ -H "Content-Type: application/json" \ -d '{"name":"Roadmap","description":"Q2 features and milestones"}'

List projects (pagination + search):

curl "http://127.0.0.1:8000/projects?limit=10&offset=0&q=road"

Partial update (only change description):

curl -X PATCH "http://127.0.0.1:8000/projects/1" \ -H "Content-Type: application/json" \ -d '{"description":"Updated: Q2 + Q3 planning"}'

Delete:

curl -X DELETE "http://127.0.0.1:8000/projects/1"

9) Practical next steps (what to add in a real app)

This foundation is intentionally small, but it’s shaped like something you can grow. If you’re taking it to production, prioritize:

• Migrations: move table creation to Alembic to avoid schema drift.

• Consistent error responses: define an error schema and reuse it across routers.

• Testing: dependency overrides let you swap get_db for a test DB session easily.

• Auth: add a dependency like get_current_user and attach it per-router or per-route.

• Performance: add indexes based on your real query patterns (e.g., search fields, foreign keys).

Once you’ve built one router like this, the rest of your API becomes copy-and-adapt: create schemas, implement CRUD with safe commits, paginate list endpoints, and keep the DB session handling consistent. That’s the difference between “it works” and “it stays maintainable.”