FastAPI in Practice: Build a Clean CRUD API with Async SQLAlchemy, Pydantic Models, and Dependency Injection

Pictalingo

·

FastAPI in Practice: Build a Clean CRUD API with Async SQLAlchemy, Pydantic Models, and Dependency Injection

FastAPI makes it easy to ship APIs quickly, but junior/mid devs often get stuck when moving from “toy examples” to a maintainable structure: async database access, request/response models, dependency injection, and tests. In this hands-on guide, you’ll build a small (but real) CRUD API for projects with:

  • FastAPI routing + validation
  • Async SQLAlchemy 2.0 sessions
  • Pydantic models for input/output (so you don’t leak DB internals)
  • Dependency injection for DB sessions
  • Basic pagination + filtering
  • Test setup with dependency overrides

The end result is a structure you can reuse for most REST backends.

Project setup

Create a virtual environment and install dependencies:

pip install fastapi uvicorn sqlalchemy aiosqlite pydantic pytest httpx

We’ll use SQLite for local development, but the patterns are the same for Postgres (swap the driver/URL).

Suggested structure:

app/ __init__.py main.py db.py models.py schemas.py crud.py routes.py tests/ test_projects.py

Database layer: async engine + session dependency

Start with app/db.py. The key is: create one engine, create a session factory, and expose a dependency that yields a session per request.

from typing import AsyncGenerator 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 while debugging SQL future=True, ) AsyncSessionLocal = sessionmaker( bind=engine, class_=AsyncSession, expire_on_commit=False, ) async def get_session() -> AsyncGenerator[AsyncSession, None]: async with AsyncSessionLocal() as session: yield session

expire_on_commit=False keeps loaded objects usable after commit (less surprising when returning responses).

SQLAlchemy model: a simple Project table

Create app/models.py:

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column from sqlalchemy import String, Boolean, DateTime, func class Base(DeclarativeBase): pass 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(String(500), default=None) is_active: Mapped[bool] = mapped_column(Boolean, default=True) created_at: Mapped[str] = mapped_column( DateTime(timezone=True), server_default=func.now(), )

This is intentionally minimal, but already includes: uniqueness, indexing, defaults, and a server-side timestamp.

Pydantic schemas: separate input from output

Create app/schemas.py. Treat DB models as internal and use schemas for request/response contracts.

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=500) class ProjectUpdate(BaseModel): name: str | None = Field(default=None, min_length=2, max_length=120) description: str | None = Field(default=None, max_length=500) is_active: bool | None = None class ProjectOut(BaseModel): id: int name: str description: str | None is_active: bool class Config: from_attributes = True # allows returning SQLAlchemy objects safely

Using ProjectOut prevents accidentally exposing fields you didn’t intend to ship.

CRUD functions: keep DB logic out of routes

Create app/crud.py. This keeps handlers thin and makes testing easier.

from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from .models import Project from .schemas import ProjectCreate, ProjectUpdate async def create_project(session: AsyncSession, data: ProjectCreate) -> Project: project = Project(name=data.name, description=data.description) session.add(project) await session.commit() await session.refresh(project) return project async def get_project(session: AsyncSession, project_id: int) -> Project | None: result = await session.execute(select(Project).where(Project.id == project_id)) return result.scalar_one_or_none() async def list_projects( session: AsyncSession, *, q: str | None = None, is_active: bool | None = None, limit: int = 20, offset: int = 0, ) -> list[Project]: stmt = select(Project).order_by(Project.id.desc()) if q: stmt = stmt.where(Project.name.ilike(f"%{q}%")) if is_active is not None: stmt = stmt.where(Project.is_active == is_active) stmt = stmt.limit(limit).offset(offset) result = await session.execute(stmt) return list(result.scalars().all()) async def update_project( session: AsyncSession, project: Project, data: ProjectUpdate ) -> Project: payload = data.model_dump(exclude_unset=True) for key, value in payload.items(): setattr(project, key, value) await session.commit() await session.refresh(project) return project async def delete_project(session: AsyncSession, project: Project) -> None: await session.delete(project) await session.commit()

Notice the pattern: select → mutate → commit → refresh. This stays consistent across most endpoints.

Routes: validation, dependencies, and good HTTP errors

Create app/routes.py:

from fastapi import APIRouter, Depends, HTTPException, Query, status from sqlalchemy.ext.asyncio import AsyncSession from .db import get_session from .schemas import ProjectCreate, ProjectUpdate, ProjectOut from . import crud router = APIRouter(prefix="/projects", tags=["projects"]) @router.post("", response_model=ProjectOut, status_code=status.HTTP_201_CREATED) async def create_project( data: ProjectCreate, session: AsyncSession = Depends(get_session), ): # Naive uniqueness handling (for real apps, catch IntegrityError too) existing = await crud.list_projects(session, q=data.name, limit=1, offset=0) if existing and existing[0].name.lower() == data.name.lower(): raise HTTPException(status_code=409, detail="Project name already exists") return await crud.create_project(session, data) @router.get("", response_model=list[ProjectOut]) async def list_projects( q: str | None = None, is_active: bool | None = None, limit: int = Query(default=20, ge=1, le=100), offset: int = Query(default=0, ge=0), session: AsyncSession = Depends(get_session), ): return await crud.list_projects(session, q=q, is_active=is_active, limit=limit, offset=offset) @router.get("/{project_id}", response_model=ProjectOut) async def get_project( project_id: int, session: AsyncSession = Depends(get_session), ): project = await crud.get_project(session, project_id) if not project: raise HTTPException(status_code=404, detail="Project not found") return project @router.patch("/{project_id}", response_model=ProjectOut) async def patch_project( project_id: int, data: ProjectUpdate, session: AsyncSession = Depends(get_session), ): project = await crud.get_project(session, project_id) if not project: raise HTTPException(status_code=404, detail="Project not found") return await crud.update_project(session, project, data) @router.delete("/{project_id}", status_code=status.HTTP_204_NO_CONTENT) async def delete_project( project_id: int, session: AsyncSession = Depends(get_session), ): project = await crud.get_project(session, project_id) if not project: raise HTTPException(status_code=404, detail="Project not found") await crud.delete_project(session, project) return None

Two practical tips you can reuse:

  • Use Query(...) constraints to prevent “accidental DoS” via huge limit values.
  • Keep DB calls in crud.py so routes read like a checklist.

Main app: wiring + creating tables

Create app/main.py. For local development, we’ll create tables on startup. (In production, use migrations, but the API structure stays the same.)

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

Run it:

uvicorn app.main:app --reload

Try requests:

curl -X POST "http://127.0.0.1:8000/projects" \ -H "Content-Type: application/json" \ -d '{"name":"Website Redesign","description":"Refresh landing pages"}'
curl "http://127.0.0.1:8000/projects?limit=10&offset=0&q=Website"

Testing: override dependencies (no real DB required)

A common FastAPI pattern is dependency overrides. For example, use an in-memory SQLite DB during tests.

Create tests/test_projects.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 get_session from app.models import Base TEST_DB_URL = "sqlite+aiosqlite:///:memory:" @pytest.fixture async def test_session(): engine = create_async_engine(TEST_DB_URL, future=True) async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) TestSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) async with TestSessionLocal() as session: yield session await engine.dispose() @pytest.fixture async def client(test_session): async def override_get_session(): yield test_session app.dependency_overrides[get_session] = override_get_session async with AsyncClient(app=app, base_url="http://test") as ac: yield ac app.dependency_overrides.clear() @pytest.mark.anyio async def test_create_and_list_projects(client): resp = await client.post("/projects", json={"name": "API", "description": "Test"}) assert resp.status_code == 201 data = resp.json() assert data["name"] == "API" resp2 = await client.get("/projects") assert resp2.status_code == 200 items = resp2.json() assert len(items) == 1 assert items[0]["name"] == "API"

Run tests:

pytest -q

Production-minded checklist (still junior-friendly)

Once this is working locally, here’s what you typically add next (in this order):

  • IntegrityError handling for unique constraints (so 409 is reliable even under race conditions)
  • Migrations (Alembic) instead of create_all in startup
  • Structured logging and request IDs
  • Auth (JWT/session) and per-route permissions
  • More robust filtering/sorting (and consistent response envelopes)

If you can build and test the CRUD API above, you’re already using the same building blocks most production FastAPI backends use—just with fewer moving parts.

Leave a Reply

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