SQLAlchemy With FastAPI

Introduction

FastAPI does not include an ORM—you pair it with SQLAlchemy 2.0 for database access. This chapter configures the engine and session, defines models, injects Session via Depends(get_db), and performs CRUD with Pydantic schemas for input and output.

Prerequisites

Install

bash
pip install sqlalchemy
pip install pymysql

Add to requirements.txt.

Database Module

app/db/session.py:

Code explanation:

  • check_same_thread=False required for SQLite with FastAPI threads
  • pool_pre_ping verifies connections before use—helps MySQL after idle timeout
  • get_db yields one session per request

Compare Flask pattern in Flask-SQLAlchemy.

Define Models

app/models/item.py:

app/models/user.py (optional relationship later):

python
from sqlalchemy.orm import Mapped, mapped_column
from sqlalchemy import String
from app.db.session 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)

Import models in app/models/__init__.py so Alembic sees metadata later.

Pydantic Schemas

app/schemas/item.py:

Create Tables (Development)

python
from app.db.session import engine, Base
from app.models import item  # noqa: F401
 
Base.metadata.create_all(bind=engine)

Run once in script or lifespan for local dev only—production uses Alembic (next chapter).

Warning

create_all Is Not for Production Teams

Use migrations for schema changes shared across environments.

CRUD Endpoints

app/api/v1/endpoints/items.py:

Code explanation:

  • db.get(Item, id) — primary key lookup (SQLAlchemy 2.0)
  • db.scalars(select(...)) — run SELECT and return ORM objects
  • db.refresh(item) — reload after commit (defaults, DB triggers)

Error Handling

python
from sqlalchemy.exc import IntegrityError
 
try:
    db.commit()
except IntegrityError:
    db.rollback()
    raise HTTPException(status_code=409, detail="Duplicate entry")

Always rollback() on failure before raising.

MySQL URI

env
DATABASE_URL=mysql+pymysql://user:password@127.0.0.1:3306/fastapi_app?charset=utf8mb4

Create database with utf8mb4—see MySQL table design.

Optional CRUD Module

Extract DB logic to app/crud/item.py functions create_item(db, ...)—keeps routes thin as apps grow.

FAQ

Session not closed?

Ensure get_db uses try/finally with db.close().

DetachedInstanceError?

Access lazy relations inside request before session closes—or eager load with selectinload.

Async SQLAlchemy?

Separate chapter—async database.

Flask-SQLAlchemy vs raw SQLAlchemy?

FastAPI docs use SQLAlchemy directly; patterns map closely to Flask-SQLAlchemy chapter.

Pagination total count?

Run separate select(func.count()) query—add in project chapters.

Relationships?

relationship() on models—see MySQL joins for SQL concepts.