Building a CRUD API with FastAPI and SQLAlchemy

Real APIs store data in a database. This lesson combines everything so far — path and query parameters, Pydantic models, response models, status codes and dependencies — with the SQLAlchemy ORM from the Database Connectivity module to build a complete CRUD API for a task tracker.

You will see the standard structure used in production FastAPI projects: an engine and a session dependency, ORM models for tables, separate Pydantic schemas for input and output, and endpoints that translate between them. The example uses SQLite, but switching to PostgreSQL or MySQL only means changing the database URL.

The Moving Parts

Engine — create_engine(DATABASE_URL), created once. SQLite needs connect_args={"check_same_thread": False} because FastAPI may use the session from another thread.
Session dependency — a yield dependency that opens a Session per request and always closes it.
ORM models — classes mapped to tables (models.py).
Schemas — Pydantic models for requests and responses (schemas.py): TaskCreate, TaskUpdate, TaskOut.
from_attributes — model_config = ConfigDict(from_attributes=True) lets a response schema read attributes straight from an ORM object.

Endpoint Patterns

Create: build an ORM object from payload.model_dump(), add, commit, refresh (to load generated ids and defaults) and return it with 201. Read one: session.get(Model, id), raising 404 when it is None. List: select() with filters, order_by, offset and limit. Update: apply model_dump(exclude_unset=True) with setattr. Delete: session.delete() and return 204. Catch IntegrityError to turn duplicate values into 409 Conflict.

Async Drivers and SQLModel

This lesson uses synchronous SQLAlchemy in plain def endpoints, which FastAPI runs in a thread pool — simple and perfectly fast for most apps. For very high concurrency you can use SQLAlchemy's asyncio extension (create_async_engine, AsyncSession) with async drivers such as asyncpg or aiosqlite in async def endpoints. SQLModel, by FastAPI's author, merges SQLAlchemy models and Pydantic schemas into one class — convenient for small projects.

Examples

A complete task-tracker CRUD API

Python

from datetime import datetime, timezone
from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException, Query, status
from fastapi.testclient import TestClient
from pydantic import BaseModel, ConfigDict, Field
from sqlalchemy import String, create_engine, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column
from sqlalchemy.pool import StaticPool

# --- database.py ---------------------------------------------------------
DATABASE_URL = "sqlite://"            # in-memory; use "sqlite:///tasks.db" or a PostgreSQL URL
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False}, poolclass=StaticPool)

class Base(DeclarativeBase):
    pass

def get_session():
    with Session(engine) as session:  # closed automatically after the response
        yield session

SessionDep = Annotated[Session, Depends(get_session)]

# --- models.py -------------------------------------------------------------
class Task(Base):
    __tablename__ = "tasks"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200), unique=True)
    priority: Mapped[int] = mapped_column(default=3)
    done: Mapped[bool] = mapped_column(default=False)
    created_at: Mapped[datetime] = mapped_column(default=lambda: datetime.now(timezone.utc))

Base.metadata.create_all(engine)

# --- schemas.py ------------------------------------------------------------
class TaskCreate(BaseModel):
    title: str = Field(min_length=1, max_length=200)
    priority: int = Field(default=3, ge=1, le=5)

class TaskUpdate(BaseModel):
    title: str | None = Field(default=None, min_length=1, max_length=200)
    priority: int | None = Field(default=None, ge=1, le=5)
    done: bool | None = None

class TaskOut(BaseModel):
    model_config = ConfigDict(from_attributes=True)   # read from ORM objects
    id: int
    title: str
    priority: int
    done: bool

# --- main.py ---------------------------------------------------------------
app = FastAPI(title="Tasks API")

def get_task_or_404(session: Session, task_id: int) -> Task:
    task = session.get(Task, task_id)
    if task is None:
        raise HTTPException(status.HTTP_404_NOT_FOUND, detail="Task not found")
    return task

@app.post("/tasks", status_code=status.HTTP_201_CREATED)
def create_task(payload: TaskCreate, session: SessionDep) -> TaskOut:
    task = Task(**payload.model_dump())
    session.add(task)
    try:
        session.commit()
    except IntegrityError:
        session.rollback()
        raise HTTPException(status.HTTP_409_CONFLICT, detail="A task with this title already exists")
    session.refresh(task)
    return task

@app.get("/tasks")
def list_tasks(session: SessionDep, done: bool | None = None,
               offset: Annotated[int, Query(ge=0)] = 0,
               limit: Annotated[int, Query(ge=1, le=100)] = 20) -> list[TaskOut]:
    query = select(Task).order_by(Task.priority, Task.id)
    if done is not None:
        query = query.where(Task.done == done)
    return session.scalars(query.offset(offset).limit(limit)).all()

@app.get("/tasks/{task_id}")
def read_task(task_id: int, session: SessionDep) -> TaskOut:
    return get_task_or_404(session, task_id)

@app.patch("/tasks/{task_id}")
def update_task(task_id: int, changes: TaskUpdate, session: SessionDep) -> TaskOut:
    task = get_task_or_404(session, task_id)
    for field, value in changes.model_dump(exclude_unset=True).items():
        setattr(task, field, value)
    session.commit()
    session.refresh(task)
    return task

@app.delete("/tasks/{task_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_task(task_id: int, session: SessionDep):
    session.delete(get_task_or_404(session, task_id))
    session.commit()

# --- trying it out ---------------------------------------------------------
client = TestClient(app)
for title, priority in [("Write report", 2), ("Review PR", 1), ("Plan sprint", 3)]:
    print(client.post("/tasks", json={"title": title, "priority": priority}).json())
print(client.post("/tasks", json={"title": "Review PR"}).status_code)
print(client.patch("/tasks/2", json={"done": True}).json())
print([t["title"] for t in client.get("/tasks").json()])
print([t["title"] for t in client.get("/tasks?done=false").json()])
print(client.delete("/tasks/3").status_code, client.get("/tasks/3").json())

Output

{'id': 1, 'title': 'Write report', 'priority': 2, 'done': False}
{'id': 2, 'title': 'Review PR', 'priority': 1, 'done': False}
{'id': 3, 'title': 'Plan sprint', 'priority': 3, 'done': False}
409
{'id': 2, 'title': 'Review PR', 'priority': 1, 'done': True}
['Review PR', 'Write report', 'Plan sprint']
['Write report', 'Plan sprint']
204 {'detail': 'Task not found'}

Switching databases is a one-line change

Python

# .env
DATABASE_URL=postgresql+psycopg://tasks_app:secret@localhost:5432/tasks

# database.py
import os
from sqlalchemy import create_engine

DATABASE_URL = os.environ.get("DATABASE_URL", "sqlite:///tasks.db")
connect_args = {"check_same_thread": False} if DATABASE_URL.startswith("sqlite") else {}
engine = create_engine(DATABASE_URL, connect_args=connect_args, pool_pre_ping=True)

Output

(no output — the rest of the application is unchanged)

Common Mistakes

Creating a global Session shared by all requests instead of one session per request via a dependency.
Returning ORM objects without from_attributes=True in the response schema.
Forgetting session.refresh() after commit and returning objects without generated ids or defaults.
Using the same Pydantic model for create, update and output.
Calling Base.metadata.create_all() to change existing tables in production — use Alembic migrations.

Key Points to Remember

Engine once, Session per request through a yield dependency.
ORM models describe tables; Pydantic schemas describe the API.
from_attributes=True lets response models read ORM objects.
CRUD: add/commit/refresh, session.get, select with filters and pagination, setattr updates, delete.
Translate database errors (IntegrityError) into proper HTTP status codes.

Practice the examples

Change an input, predict the result, then compare it with the output. Explain why the result changes.

Use your local project environment for these examples. Codelab currently runs Python and HTML/CSS/JavaScript; framework examples may need project dependencies.