🟩 FastAPI

Databases with SQLAlchemy

📖 Lesson 41 ⏱ 55 min 🧪 5 questions 💻 3 exercises

🎯 Learning Objectives

Understand the SQLAlchemy Core vs ORM distinction
Define mapped classes with DeclarativeBase and column types
Create async engines and session factories with asyncpg
Perform CRUD operations using AsyncSession and select()
Define relationships: one-to-many, many-to-many with association tables
Write and run Alembic migrations
Integrate the async session as a FastAPI yield dependency

SQLAlchemy Overview

SQLAlchemy ships two layers that you can use together or independently:

Core — a SQL expression language that sits close to raw SQL but gives you Python composability and cross-database portability.
ORM — maps Python classes to database tables, tracks object state, and manages relationships.

SQLAlchemy 2.0 unified the API — the ORM now uses the same select() style as Core, eliminating the old session.query() split. Async support is first-class via asyncpg (PostgreSQL) or aiosqlite (SQLite for dev/testing).

pip install sqlalchemy asyncpg alembic
# For dev/testing with SQLite:
pip install aiosqlite

terminal

Approach	When to use
Raw SQL (`text()`)	Complex queries, performance-critical, legacy schemas
Core (`select()`)	Fine-grained control without full ORM overhead
ORM (`Session` + mapped classes)	Standard CRUD, relationships, most FastAPI apps
Alembic	Schema migrations in all three cases

SQLAlchemy 2.0 introduced a clean select(Model) ORM style that works identically in sync and async contexts. Avoid the legacy session.query(Model) API in new code.

Defining Models

SQLAlchemy 2.0 uses Mapped[T] type annotations together with mapped_column() to declare columns. This gives you full type-checker support and a single source of truth for both the schema and the Python type.

# models.py
from sqlalchemy import String, Integer, Float, Boolean, ForeignKey, DateTime, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
from datetime import datetime


class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "users"

    id:         Mapped[int]      = mapped_column(Integer, primary_key=True, index=True)
    username:   Mapped[str]      = mapped_column(String(50), unique=True, nullable=False, index=True)
    email:      Mapped[str]      = mapped_column(String(255), unique=True, nullable=False)
    hashed_pw:  Mapped[str]      = mapped_column(String(255), nullable=False)
    is_active:  Mapped[bool]     = mapped_column(Boolean, default=True)
    created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())

    posts: Mapped[list["Post"]] = relationship(
        "Post", back_populates="author", cascade="all, delete-orphan"
    )

    def __repr__(self) -> str:
        return f"<User id={self.id} username={self.username!r}>"


class Post(Base):
    __tablename__ = "posts"

    id:         Mapped[int]  = mapped_column(Integer, primary_key=True, index=True)
    title:      Mapped[str]  = mapped_column(String(200), nullable=False)
    body:       Mapped[str]  = mapped_column(String, nullable=False)
    published:  Mapped[bool] = mapped_column(Boolean, default=False)
    author_id:  Mapped[int]  = mapped_column(ForeignKey("users.id"), nullable=False)

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

models.py

Key points:

Mapped[T] + mapped_column() is the SQLAlchemy 2.0 typed style — it replaces the old Column() declarations.
server_default=func.now() generates a DEFAULT NOW() clause in the DDL — the database sets the timestamp, not Python.
cascade="all, delete-orphan" means deleting a User automatically deletes their Post rows.
relationship() with back_populates creates a bidirectional link — user.posts and post.author stay in sync.

Async Engine & Session Factory

# database.py
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from models import Base

# PostgreSQL in production
DATABASE_URL = "postgresql+asyncpg://user:password@localhost:5432/mydb"

# SQLite for local dev / tests
# DATABASE_URL = "sqlite+aiosqlite:///./dev.db"

engine = create_async_engine(
    DATABASE_URL,
    echo=True,          # log all SQL (disable in production)
    pool_size=10,
    max_overflow=20,
)

AsyncSessionLocal = async_sessionmaker(
    engine,
    expire_on_commit=False,   # keep objects usable after commit
    class_=AsyncSession,
)


async def create_tables():
    """Create all tables — dev only; use Alembic in production."""
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

database.py

expire_on_commit=False is essential for async sessions. Without it, accessing any attribute after commit() triggers a lazy load — which fails in an async context because SQLAlchemy cannot implicitly issue a synchronous I/O call. Always set this to False for async sessions.

Use echo=True during development to see every SQL query in the console. Disable it in production to avoid log noise and potential credential leakage.

FastAPI Session Dependency

Expose the async session via a FastAPI yield dependency. The dependency commits on success or rolls back on exception — route handlers never call commit() themselves.

# dependencies.py
from collections.abc import AsyncGenerator
from sqlalchemy.ext.asyncio import AsyncSession
from database import AsyncSessionLocal


async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

dependencies.py

Using the dependency in a route:

from fastapi import FastAPI, Depends
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from dependencies import get_db
from models import User

app = FastAPI()


@app.get("/users")
async def list_users(db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(User).where(User.is_active == True))
    return result.scalars().all()

main.py

CRUD Operations

# crud.py
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, update, delete
from models import User, Post


# ── CREATE ──
async def create_user(db: AsyncSession, username: str, email: str, hashed_pw: str) -> User:
    user = User(username=username, email=email, hashed_pw=hashed_pw)
    db.add(user)
    await db.flush()   # assign id without committing (commit handled by dependency)
    return user


# ── READ ──
async def get_user_by_id(db: AsyncSession, user_id: int) -> User | None:
    return await db.get(User, user_id)   # fastest single-PK lookup


async def get_user_by_email(db: AsyncSession, email: str) -> User | None:
    result = await db.execute(select(User).where(User.email == email))
    return result.scalar_one_or_none()


async def list_users(db: AsyncSession, skip: int = 0, limit: int = 10) -> list[User]:
    result = await db.execute(
        select(User).offset(skip).limit(limit).order_by(User.id)
    )
    return list(result.scalars().all())


# ── UPDATE ──
async def update_user_email(db: AsyncSession, user_id: int, new_email: str) -> User | None:
    user = await db.get(User, user_id)
    if not user:
        return None
    user.email = new_email           # modify the mapped attribute directly
    await db.flush()
    return user


# ── DELETE ──
async def delete_user(db: AsyncSession, user_id: int) -> bool:
    user = await db.get(User, user_id)
    if not user:
        return False
    await db.delete(user)
    return True

crud.py

flush() vs commit(): flush() writes changes to the database within the current transaction (assigning auto-generated IDs) but does not finalise. commit() finalises the transaction. Since our get_db dependency commits after yield, CRUD functions only need flush() to get IDs back.

Method	Returns	When to use
`scalar_one_or_none()`	One object or `None`	Lookup by unique field
`scalar_one()`	One object (raises if 0 or 2+)	When exactly one row is expected
`scalars().all()`	List of objects	Multiple rows

Relationships & Eager Loading

In async SQLAlchemy you must explicitly load relationships using eager-loading options — lazy loading is not supported in async context.

from sqlalchemy.orm import selectinload, joinedload
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from models import User, Post


# ── selectinload — separate IN query (recommended for one-to-many) ──
async def get_user_with_posts(db: AsyncSession, user_id: int) -> User | None:
    result = await db.execute(
        select(User)
        .where(User.id == user_id)
        .options(selectinload(User.posts))   # loads posts in a second query
    )
    return result.scalar_one_or_none()


# ── joinedload — single JOIN query (good for many-to-one / single object) ──
async def get_post_with_author(db: AsyncSession, post_id: int) -> Post | None:
    result = await db.execute(
        select(Post)
        .where(Post.id == post_id)
        .options(joinedload(Post.author))    # JOIN users ON posts.author_id = users.id
    )
    return result.scalar_one_or_none()

crud.py

Many-to-many relationships use an association table:

from sqlalchemy import Table, Column, ForeignKey, Integer, String
from sqlalchemy.orm import Mapped, mapped_column, relationship
from models import Base

post_tags = Table(
    "post_tags", Base.metadata,
    Column("post_id", ForeignKey("posts.id"), primary_key=True),
    Column("tag_id",  ForeignKey("tags.id"),  primary_key=True),
)


class Tag(Base):
    __tablename__ = "tags"

    id:   Mapped[int] = mapped_column(Integer, primary_key=True)
    name: Mapped[str] = mapped_column(String(50), unique=True)

    posts: Mapped[list["Post"]] = relationship(
        "Post", secondary=post_tags, back_populates="tags"
    )

models.py (addition)

Add the corresponding tags relationship to the Post model:

# Add to the Post class
tags: Mapped[list["Tag"]] = relationship(
    "Tag", secondary=post_tags, back_populates="posts"
)

models.py

In async SQLAlchemy you must always explicitly load relationships using selectinload or joinedload — lazy loading is not supported in async context (it would silently trigger a sync DB call and raise MissingGreenlet).

Alembic Migrations

Alembic is the official migration tool for SQLAlchemy. It tracks schema changes as versioned migration scripts so you can upgrade, downgrade, and reproduce your database schema deterministically.

# 1 — Initialise Alembic
pip install alembic
alembic init alembic

terminal

Modify alembic/env.py to support async engines:

# alembic/env.py — key changes
import asyncio
from sqlalchemy.ext.asyncio import create_async_engine
from alembic import context
from models import Base
from database import DATABASE_URL

target_metadata = Base.metadata  # tell Alembic what schema to compare against


def do_run_migrations(connection):
    context.configure(connection=connection, target_metadata=target_metadata)
    with context.begin_transaction():
        context.run_migrations()


async def run_async_migrations():
    engine = create_async_engine(DATABASE_URL)
    async with engine.begin() as conn:
        await conn.run_sync(do_run_migrations)
    await engine.dispose()


def run_migrations_online():
    asyncio.run(run_async_migrations())

alembic/env.py

Common Alembic commands:

# Auto-generate migration from model changes
alembic revision --autogenerate -m "add users and posts tables"

# Apply pending migrations
alembic upgrade head

# Roll back one step
alembic downgrade -1

# View migration history
alembic history --verbose

# Show current revision in DB
alembic current

terminal

--autogenerate detects column additions/removals and index changes but does NOT detect: column renames, changes to server_default, or custom SQL types. Always review the generated migration file before applying.

N+1 Problem & Query Optimisation

The N+1 problem occurs when you load N rows then issue a separate query for each row's related data — 1 query for users + N queries for their posts = N+1 total.

Bad — N+1 with a loop

# ❌ N+1: 1 query for users + N queries for posts
result = await db.execute(select(User))
users = result.scalars().all()
for user in users:
    posts_result = await db.execute(select(Post).where(Post.author_id == user.id))
    user_posts = posts_result.scalars().all()   # 1 extra query PER user!

bad_n_plus_1.py

Good — eager loading in 2 queries

# ✅ 2 queries total: one for users, one IN-query for all their posts
from sqlalchemy.orm import selectinload

result = await db.execute(
    select(User).options(selectinload(User.posts))
)
users = result.scalars().all()
# user.posts already loaded — no extra queries

good_eager_load.py

Pagination — offset vs keyset

# Offset pagination — simple but slow on large tables
result = await db.execute(
    select(Post).order_by(Post.id).offset(200).limit(20)
)

# Keyset (cursor) pagination — fast at any depth
last_seen_id = 200
result = await db.execute(
    select(Post).where(Post.id > last_seen_id).order_by(Post.id).limit(20)
)

pagination.py

Aggregations with `func`

from sqlalchemy import func

result = await db.execute(
    select(User.id, func.count(Post.id).label("post_count"))
    .join(Post, isouter=True)
    .group_by(User.id)
    .order_by(func.count(Post.id).desc())
    .limit(10)
)
rows = result.all()   # list of Row(id, post_count)

aggregations.py

Use EXPLAIN ANALYZE in PostgreSQL to inspect query plans. SQLAlchemy's echo=True shows emitted SQL but not the execution plan — paste the query into psql or pgAdmin for real optimisation.

Pydantic + SQLAlchemy Integration

Keep Pydantic schemas (request/response shapes) separate from SQLAlchemy models (DB tables). The bridge is from_attributes=True — Pydantic reads ORM object attributes instead of requiring a dict.

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

class UserCreate(BaseModel):
    username: str
    email: EmailStr
    password: str

class UserOut(BaseModel):
    model_config = ConfigDict(from_attributes=True)   # read from ORM object
    id: int
    username: str
    email: str
    is_active: bool
    created_at: datetime

class PostOut(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: int
    title: str
    published: bool
    author: UserOut           # nested ORM → Pydantic conversion

schemas.py

# Route that returns an ORM object directly
@app.post("/users", response_model=UserOut, status_code=201)
async def create_user_route(
    body: UserCreate,
    db: AsyncSession = Depends(get_db),
):
    existing = await get_user_by_email(db, body.email)
    if existing:
        raise HTTPException(409, detail="Email already registered")
    user = await create_user(db, body.username, body.email, hash_password(body.password))
    return user   # FastAPI calls UserOut.model_validate(user) via from_attributes=True

routes.py

from_attributes=True (formerly orm_mode = True in Pydantic v1) enables UserOut.model_validate(orm_obj) to read attributes instead of dict keys. FastAPI calls this automatically when serialising a response_model, so you can return raw ORM objects from route handlers without manual conversion.

Testing with a Real SQLite DB

Use SQLite + aiosqlite for integration tests — it exercises the full async SQLAlchemy stack without needing a running PostgreSQL instance in CI.

# conftest.py
import pytest, pytest_asyncio
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from fastapi.testclient import TestClient
from main import app
from database import get_db
from models import Base

TEST_DB_URL = "sqlite+aiosqlite:///./test.db"

@pytest_asyncio.fixture(scope="session")
async def engine():
    eng = create_async_engine(TEST_DB_URL)
    async with eng.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield eng
    async with eng.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
    await eng.dispose()

@pytest_asyncio.fixture
async def db_session(engine):
    AsyncTestSession = async_sessionmaker(engine, expire_on_commit=False)
    async with AsyncTestSession() as session:
        yield session
        await session.rollback()   # undo test changes — fast, no DDL

@pytest.fixture
def client(db_session):
    async def override_get_db():
        yield db_session
    app.dependency_overrides[get_db] = override_get_db
    yield TestClient(app)
    app.dependency_overrides.clear()

conftest.py

Rollback after each test instead of dropping/recreating tables — much faster. Use scope="session" for the engine (created once) and default scope="function" for the session (fresh per test).

Best Practices

Use async SQLAlchemy for FastAPI — sync sessions block the event loop.
Always load relationships explicitly — selectinload for one-to-many, joinedload for many-to-one. Never rely on lazy loading in async code.
Use Alembic for all schema changes — never call Base.metadata.create_all() in production.
flush() in CRUD, commit() in the dependency — keeps transaction control at the boundary.
Set expire_on_commit=False on async sessions — prevents DetachedInstanceError.
Use from_attributes=True on all response Pydantic models — cleanly converts ORM objects without manual .model_dump().
Add indexes on foreign keys and frequently queried columns — mapped_column(index=True) generates a B-tree index.
Test with SQLite + aiosqlite in CI — fast, zero-config, schema-compatible with PostgreSQL for most queries.

Exercises

Exercise 1 — Blog CRUD API

Build a complete FastAPI + SQLAlchemy async blog API:

Models: User (id, username, email, hashed_pw) and Post (id, title, body, published, author_id → User)
CRUD functions: create_user, get_user, create_post, list_posts(published_only), publish_post
Routes: POST /users, POST /posts, GET /posts, GET /posts/{id}, PATCH /posts/{id}/publish
Response models with from_attributes=True; raise 404 on missing resources

💡 Hint — list posts with author pre-loaded

@app.get("/posts", response_model=list[PostOut])
async def list_posts(published_only: bool = False, db: AsyncSession = Depends(get_db)):
    q = select(Post).options(selectinload(Post.author)).order_by(Post.id)
    if published_only:
        q = q.where(Post.published == True)
    result = await db.execute(q)
    return result.scalars().all()

Exercise 2 — Tags (Many-to-Many)

Extend the blog with a Tag system:

Add Tag model and post_tags association table
POST /tags — create a tag
POST /posts/{id}/tags — attach a list of tag names (create missing tags)
GET /posts/{id} — include tags in the response

💡 Hint — attach tags

async def add_tags_to_post(db: AsyncSession, post_id: int, tag_names: list[str]):
    result = await db.execute(
        select(Post).where(Post.id == post_id).options(selectinload(Post.tags))
    )
    post = result.scalar_one_or_none()
    if not post:
        return None
    for name in tag_names:
        tag_r = await db.execute(select(Tag).where(Tag.name == name))
        tag = tag_r.scalar_one_or_none()
        if not tag:
            tag = Tag(name=name)
            db.add(tag)
            await db.flush()
        if tag not in post.tags:
            post.tags.append(tag)
    return post

Exercise 3 — Write & Run an Alembic Migration

Practice the full Alembic workflow:

Start from the Blog schema (Exercise 1)
alembic init alembic — configure env.py with your models and async engine
alembic revision --autogenerate -m "initial schema" — inspect the generated file
alembic upgrade head — confirm tables exist
Add bio: Mapped[str | None] to User
alembic revision --autogenerate -m "add user bio" then alembic upgrade head
alembic downgrade -1 — confirm the column is removed

💡 Hint

alembic current          # show active revision
alembic history --verbose   # show full migration log

Databases with SQLAlchemy

🎯 Learning Objectives

SQLAlchemy Overview

Defining Models

Async Engine & Session Factory

FastAPI Session Dependency

CRUD Operations

Relationships & Eager Loading

Alembic Migrations

N+1 Problem & Query Optimisation

Bad — N+1 with a loop

Good — eager loading in 2 queries

Pagination — offset vs keyset

Aggregations with func

Pydantic + SQLAlchemy Integration

Testing with a Real SQLite DB

Best Practices

Exercises

Exercise 1 — Blog CRUD API

Exercise 2 — Tags (Many-to-Many)

Exercise 3 — Write & Run an Alembic Migration

Aggregations with `func`