FastAPI in Practice: Build a Production-Ready CRUD API with Async SQLAlchemy, Pydantic v2, and Real Tests

Pictalingo

·

FastAPI in Practice: Build a Production-Ready CRUD API with Async SQLAlchemy, Pydantic v2, and Real Tests

FastAPI is great for getting an API running quickly—but the “hello world” examples don’t show the patterns you’ll need once you add a database, validation, and tests. This hands-on guide walks you through a small but realistic CRUD service using:

  • FastAPI for HTTP routing + dependency injection
  • SQLAlchemy 2.x async ORM for database access
  • Pydantic v2 for request/response models
  • pytest for tests that hit real endpoints

We’ll build a “Tasks” API (create/list/get/update/delete). The patterns transfer directly to bigger projects.

Project Structure (Simple, Maintainable)

Here’s a minimal layout that avoids “everything in main.py”:

app/ __init__.py main.py db.py models.py schemas.py crud.py deps.py tests/ test_tasks.py

Install Dependencies

Use a virtual environment, then:

pip install fastapi "uvicorn[standard]" sqlalchemy aiosqlite pydantic pytest httpx

We’ll use SQLite for the demo (aiosqlite), but the code pattern works the same with Postgres (swap the driver/URL).

Database Setup (Async SQLAlchemy 2.x)

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 while debugging SQL future=True, ) async_session_factory = sessionmaker( engine, class_=AsyncSession, expire_on_commit=False, ) class Base(DeclarativeBase): pass

Key details:

  • expire_on_commit=False keeps attributes accessible after commit (less surprise for juniors).
  • The async engine + AsyncSession prevents blocking under load.

Define a Model

Create app/models.py:

from sqlalchemy import String, Boolean from sqlalchemy.orm import Mapped, mapped_column from .db import Base class Task(Base): __tablename__ = "tasks" id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True) title: Mapped[str] = mapped_column(String(200), nullable=False) done: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False)

Pydantic Schemas (Pydantic v2)

Create app/schemas.py:

from pydantic import BaseModel, ConfigDict, Field 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 TaskRead(BaseModel): model_config = ConfigDict(from_attributes=True) id: int title: str done: bool

from_attributes=True lets Pydantic build responses directly from ORM objects.

Dependency: Get a DB Session Per Request

Create app/deps.py:

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

This gives you a clean, testable way to inject a session in endpoints.

CRUD Functions (Keep DB Logic Out of Routes)

Create app/crud.py:

from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from .models import Task async def create_task(db: AsyncSession, title: str) -> Task: task = Task(title=title, done=False) db.add(task) await db.commit() await db.refresh(task) return task async def list_tasks(db: AsyncSession, limit: int = 50, offset: int = 0) -> list[Task]: stmt = select(Task).order_by(Task.id.desc()).limit(limit).offset(offset) result = await db.execute(stmt) return list(result.scalars().all()) async def get_task(db: AsyncSession, task_id: int) -> Task | None: stmt = select(Task).where(Task.id == task_id) result = await db.execute(stmt) return result.scalars().first() async def update_task( db: AsyncSession, task: Task, title: str | None = None, done: bool | None = None, ) -> Task: if title is not None: task.title = title if done is not None: task.done = done await db.commit() await db.refresh(task) return task async def delete_task(db: AsyncSession, task: Task) -> None: await db.delete(task) await db.commit()

Why this matters: routes stay readable, and CRUD functions become easy to unit test later.

FastAPI App + Routes

Create app/main.py:

from fastapi import FastAPI, Depends, HTTPException, status from sqlalchemy.ext.asyncio import AsyncSession from .db import engine, Base from .deps import get_db from . import crud, schemas app = FastAPI(title="Tasks API") @app.on_event("startup") async def on_startup() -> None: # Simple auto-migration for demo purposes (don’t use this as a real migration system) async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) @app.post("/tasks", response_model=schemas.TaskRead, status_code=status.HTTP_201_CREATED) async def create_task(payload: schemas.TaskCreate, db: AsyncSession = Depends(get_db)): task = await crud.create_task(db, title=payload.title) return task @app.get("/tasks", response_model=list[schemas.TaskRead]) async def list_tasks( limit: int = 50, offset: int = 0, db: AsyncSession = Depends(get_db), ): # Guardrails for juniors: avoid “limit=1000000” accidents limit = min(max(limit, 1), 100) offset = max(offset, 0) return await crud.list_tasks(db, limit=limit, offset=offset) @app.get("/tasks/{task_id}", response_model=schemas.TaskRead) async def get_task(task_id: int, db: AsyncSession = Depends(get_db)): task = await crud.get_task(db, task_id=task_id) if not task: raise HTTPException(status_code=404, detail="Task not found") return task @app.patch("/tasks/{task_id}", response_model=schemas.TaskRead) async def patch_task(task_id: int, payload: schemas.TaskUpdate, db: AsyncSession = Depends(get_db)): task = await crud.get_task(db, task_id=task_id) if not task: raise HTTPException(status_code=404, detail="Task not found") return await crud.update_task(db, task, title=payload.title, done=payload.done) @app.delete("/tasks/{task_id}", status_code=status.HTTP_204_NO_CONTENT) async def delete_task(task_id: int, db: AsyncSession = Depends(get_db)): task = await crud.get_task(db, task_id=task_id) if not task: raise HTTPException(status_code=404, detail="Task not found") await crud.delete_task(db, task) return None

Run it:

uvicorn app.main:app --reload

Then open /docs to try it in Swagger UI.

Hands-On: Try the API with cURL

# Create curl -X POST http://127.0.0.1:8000/tasks \ -H "Content-Type: application/json" \ -d '{"title":"Ship the feature"}' # List curl http://127.0.0.1:8000/tasks # Update curl -X PATCH http://127.0.0.1:8000/tasks/1 \ -H "Content-Type: application/json" \ -d '{"done":true}' # Delete curl -X DELETE http://127.0.0.1:8000/tasks/1

Testing the API (Without Spinning Up a Server)

Good tests are what separate “works on my machine” from “we can deploy this safely.” We’ll use httpx to call the app in-process.

Create tests/test_tasks.py:

import pytest from httpx import AsyncClient from app.main import app from app.db import engine, Base @pytest.fixture(autouse=True, scope="function") async def fresh_db(): # Start each test with clean tables (simple demo approach) async with engine.begin() as conn: await conn.run_sync(Base.metadata.drop_all) await conn.run_sync(Base.metadata.create_all) yield @pytest.mark.anyio async def test_task_lifecycle(): 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 task = r.json() assert task["id"] == 1 assert task["done"] is False # List r = await client.get("/tasks") assert r.status_code == 200 assert len(r.json()) == 1 # Patch r = await client.patch("/tasks/1", json={"done": True}) assert r.status_code == 200 assert r.json()["done"] is True # Delete r = await client.delete("/tasks/1") assert r.status_code == 204 # Get should 404 r = await client.get("/tasks/1") assert r.status_code == 404

Run tests:

pytest -q

Practical Tips You’ll Reuse in Real Projects

  • Keep validation in schemas, not route code. Let TaskCreate/TaskUpdate enforce constraints. It keeps endpoints clean and consistent.

  • Separate CRUD from routes. Your future self will thank you when you add pagination, filters, or move logic into services.

  • Clamp user-controlled pagination. The min/max guardrails prevent accidental abuse and surprise load spikes.

  • Prefer PATCH for partial updates. It maps nicely to optional fields like done and title.

  • Don’t treat “auto create tables” as migrations. For production, use a migration tool (e.g., Alembic) and apply migrations in CI/CD.

Where to Go Next

Once this is working, the next upgrades that matter most:

  • Config via environment variables (database URL, log level) using pydantic-settings.

  • Better error shapes (consistent API errors, request IDs).

  • Alembic migrations (schema changes without dropping tables).

  • Auth (JWT or session-based) and role checks via dependencies.

  • Structured logging and tracing if you’re going microservices.

You now have a clean FastAPI baseline: async DB access, strong validation, and tests that prove your endpoints work. From here, you can scale features without turning your codebase into a single-file experiment.

Leave a Reply

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