FastAPI + SQLAlchemy: Your First Database-Backed API

The first version of any real API talks to a database. This post wires FastAPI to SQLAlchemy 2.0 — the modern, fully-typed version of Python’s most popular ORM — and walks through a complete CRUD example. We’ll keep things synchronous to focus on the moving parts; an async variant uses the same shape with AsyncSession.

What you need installed

# Core dependencies for this post.
pip install "fastapi[standard]" sqlalchemy pydantic
# Optional but used at the end:
pip install alembic

We’ll use SQLite for the example so there’s nothing to spin up. Swapping to Postgres later is a one-line URL change.

Engine, session, and base class

SQLAlchemy splits its world into three pieces you set up once:

# db.py
from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, sessionmaker

# The engine owns the connection pool. One per app.
engine = create_engine(
    "sqlite:///./app.db",
    echo=False,                  # set True to see SQL during development
    future=True,
    connect_args={"check_same_thread": False},  # SQLite-only
)

# sessionmaker is a factory for Sessions — short-lived units of work.
SessionLocal = sessionmaker(
    bind=engine,
    autoflush=False,
    autocommit=False,
    future=True,
)

# Every model inherits from this. SQLAlchemy 2.0 uses DeclarativeBase.
class Base(DeclarativeBase):
    pass

Three things to know:

• The engine is global and shared. It manages the underlying database connections.
• A session is a short-lived workspace for a unit of work. You get one per request and discard it.
• Base is the registry every model declares against.

Defining typed models

SQLAlchemy 2.0 introduced Mapped[T] and mapped_column(...), which give you full type checking on your models:

# models.py
from datetime import datetime
from sqlalchemy import String, ForeignKey
from sqlalchemy.orm import Mapped, mapped_column, relationship
from .db import Base

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(255), unique=True, index=True)
    name: Mapped[str] = mapped_column(String(100))
    created_at: Mapped[datetime] = mapped_column(default=datetime.utcnow)

    posts: Mapped[list["Post"]] = relationship(back_populates="author")

class Post(Base):
    __tablename__ = "posts"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    body: Mapped[str] = mapped_column(String(10_000))
    author_id: Mapped[int] = mapped_column(ForeignKey("users.id"))

    author: Mapped["User"] = relationship(back_populates="posts")

A few details:

• Mapped[int] is the static type. mapped_column(...) adds runtime/schema metadata.
• A column typed Mapped[str | None] (or Optional[str]) is nullable; otherwise it’s NOT NULL.
• relationship describes the Python-level link between two model classes.
• back_populates ties the two sides together so updates propagate both ways.

Create the tables once (Alembic handles this in production, but for prototyping you can call):

from .db import engine, Base
Base.metadata.create_all(bind=engine)

The get_db dependency

The pattern from the dependencies post returns:

# deps.py
from typing import Annotated, Iterator
from fastapi import Depends
from sqlalchemy.orm import Session
from .db import SessionLocal

def get_db() -> Iterator[Session]:
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

DB = Annotated[Session, Depends(get_db)]

Every route that touches the database takes db: DB. The session opens when the request arrives and closes after the response is sent.

Pydantic schemas separate from ORM models

A common beginner mistake is to expose SQLAlchemy models directly. Don’t — keep request/response shapes in Pydantic and the storage shape in SQLAlchemy. They will diverge over time.

# schemas.py
from datetime import datetime
from pydantic import BaseModel, ConfigDict, EmailStr, Field
from typing import Annotated

class UserCreate(BaseModel):
    email: EmailStr
    name: Annotated[str, Field(min_length=1, max_length=100)]

class UserOut(BaseModel):
    model_config = ConfigDict(from_attributes=True)  # read attributes off ORM objects

    id: int
    email: EmailStr
    name: str
    created_at: datetime

from_attributes=True is the v2 setting that lets Pydantic build a model from an arbitrary object with matching attributes — exactly what we want for an ORM row.

CRUD endpoints

A full app file, top to bottom:

# main.py
from fastapi import FastAPI, HTTPException, status
from sqlalchemy import select

from .db import Base, engine
from .deps import DB
from .models import User
from .schemas import UserCreate, UserOut

# For prototyping; production uses Alembic.
Base.metadata.create_all(bind=engine)

app = FastAPI(title="Users API")

@app.post("/users", response_model=UserOut, status_code=status.HTTP_201_CREATED)
def create_user(payload: UserCreate, db: DB) -> User:
    # Check uniqueness explicitly for a cleaner error than the DB IntegrityError.
    existing = db.scalar(select(User).where(User.email == payload.email))
    if existing is not None:
        raise HTTPException(status_code=409, detail="Email already registered")

    user = User(email=payload.email, name=payload.name)
    db.add(user)
    db.commit()
    db.refresh(user)        # reload id and defaults from the DB
    return user

@app.get("/users", response_model=list[UserOut])
def list_users(db: DB, limit: int = 50, offset: int = 0) -> list[User]:
    stmt = select(User).order_by(User.id).limit(limit).offset(offset)
    return list(db.scalars(stmt))

@app.get("/users/{user_id}", response_model=UserOut)
def read_user(user_id: int, db: DB) -> User:
    user = db.get(User, user_id)
    if user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return user

@app.delete("/users/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_user(user_id: int, db: DB) -> None:
    user = db.get(User, user_id)
    if user is None:
        raise HTTPException(status_code=404, detail="User not found")
    db.delete(user)
    db.commit()

Run it:

fastapi dev main.py

POST a few users at /docs, list them, fetch one by ID, delete one. The whole CRUD surface in under 50 lines of route code, plus a clear OpenAPI page.

Update with a PATCH

Partial updates use Pydantic’s exclude_unset to know which fields the caller actually sent:

class UserUpdate(BaseModel):
    email: EmailStr | None = None
    name: Annotated[str, Field(min_length=1, max_length=100)] | None = None

@app.patch("/users/{user_id}", response_model=UserOut)
def update_user(user_id: int, payload: UserUpdate, db: DB) -> User:
    user = db.get(User, user_id)
    if user is None:
        raise HTTPException(status_code=404, detail="User not found")

    changes = payload.model_dump(exclude_unset=True)
    for field, value in changes.items():
        setattr(user, field, value)

    db.commit()
    db.refresh(user)
    return user

model_dump(exclude_unset=True) is the canonical “only the fields the client sent” idiom. The loop applies them; SQLAlchemy tracks which columns changed and emits a minimal UPDATE.

A relationship in action

Add a Post endpoint that uses the relationship:

class PostCreate(BaseModel):
    title: str
    body: str

class PostOut(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: int
    title: str
    body: str
    author_id: int

@app.post("/users/{user_id}/posts", response_model=PostOut, status_code=201)
def create_post(user_id: int, payload: PostCreate, db: DB):
    author = db.get(User, user_id)
    if author is None:
        raise HTTPException(status_code=404, detail="User not found")

    post = Post(title=payload.title, body=payload.body, author=author)
    db.add(post)
    db.commit()
    db.refresh(post)
    return post

@app.get("/users/{user_id}/posts", response_model=list[PostOut])
def list_user_posts(user_id: int, db: DB):
    user = db.get(User, user_id)
    if user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return user.posts        # SQLAlchemy issues a SELECT for the relationship

The relationship is read like a normal Python list. Internally SQLAlchemy issues the right query (and you can opt into eager loading with selectinload to avoid N+1 problems in larger projects).

Common pitfalls

• Forgetting db.commit(). Without it, your changes vanish when the session closes.
• Returning a session-bound object after the session closed. With response_model plus from_attributes=True, FastAPI reads the attributes before the session ends, so you’re safe. If you return a raw model and access it later, you can hit DetachedInstanceError.
• expire_on_commit=True (the default). After commit(), every loaded attribute is expired and the next read re-queries it. db.refresh(obj) is the explicit way to reload; or set expire_on_commit=False on the sessionmaker if you prefer the snapshot to stay valid.
• Sharing a session across requests. Don’t. The per-request get_db dependency keeps you honest.

Where Alembic fits

Base.metadata.create_all(...) is fine for the first day. As your schema evolves, you want migrations — versioned scripts that bring a database from version N to N+1 reversibly. That’s Alembic.

A typical setup:

# Inside your project.
alembic init alembic

Edit alembic.ini and alembic/env.py to point at your engine and the Base.metadata. Then iterate:

# Detect the diff between current DB and your models, write a migration script.
alembic revision --autogenerate -m "create users and posts"

# Apply pending migrations.
alembic upgrade head

# Roll back one step.
alembic downgrade -1

Two rules that save real teams pain:

• Review autogenerated migrations by hand. Alembic’s diff is usually right but not always — particularly with renames and constraint changes.
• Never edit a migration that has already shipped. Add a new one that undoes the change.

For first-day prototypes, skip Alembic. For anything you ship, set it up before you need it.

Async variant in 10 lines

The whole shape carries to async — replace the engine, sessionmaker, dependency, and route signatures:

from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine

engine = create_async_engine("postgresql+asyncpg://localhost/app")
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)

async def get_db() -> "AsyncIterator[AsyncSession]":
    async with AsyncSessionLocal() as db:
        yield db

DB = Annotated[AsyncSession, Depends(get_db)]

@app.get("/users/{user_id}", response_model=UserOut)
async def read_user(user_id: int, db: DB):
    user = await db.get(User, user_id)
    if user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return user

Same model layer, async session, await on every database call. SQLite supports async via aiosqlite; Postgres uses asyncpg; MySQL uses aiomysql.

Recap

You now know:

• SQLAlchemy 2.0 models use Mapped[T] plus mapped_column(...) for fully typed columns.
• The engine is global; sessions are short-lived units of work created via sessionmaker.
• A get_db generator dependency gives every request its own session and closes it cleanly.
• Pydantic schemas (UserCreate, UserOut) stay separate from ORM models; from_attributes=True bridges them.
• model_dump(exclude_unset=True) is the canonical pattern for PATCH endpoints.
• Relationships are first-class Python attributes; selectinload solves N+1 problems when they appear.
• Alembic handles versioned migrations in production; create_all is fine for the first day only.

Next steps

From here, the path forward is the rest of a real service — authentication with JWT, caching with Redis, background work with a queue, and deploying with Docker. The patterns in this post compose with all of them.

Related: FastAPI Async and Dependencies, FastAPI Routes and Pydantic Models, What Is FastAPI?.

Questions or feedback? Email codeloomdevv@gmail.com.