SQLAlchemy Setup

SQLAlchemy 2.x introduced a cleaner API with first-class async support. Using it with FastAPI requires a specific setup: an async engine, a session factory, and a get_db dependency that handles session lifecycle per request.

Learning Focus

By the end of this lesson you can: create an async SQLAlchemy engine, define ORM models with the declarative base, write a session factory, and wire the get_db dependency into FastAPI.

Installation

install-sqlalchemy.sh

pip install "sqlalchemy[asyncio]>=2.0" asyncpg psycopg2-binary
# asyncpg — async PostgreSQL driver (preferred for async FastAPI)
# psycopg2 — sync driver (for Alembic and CLI tools)

Async Engine and Session Factory

app/db/session.py

from sqlalchemy.ext.asyncio import (
    AsyncSession,
    create_async_engine,
    async_sessionmaker,
)
from app.core.config import settings

# Create the async engine
engine = create_async_engine(
    settings.DATABASE_URL,        # e.g. postgresql+asyncpg://user:pass@host/db
    echo=settings.DEBUG,          # Log SQL queries in debug mode
    pool_size=10,                 # Connection pool size
    max_overflow=20,              # Extra connections beyond pool_size
    pool_pre_ping=True,           # Test connections before use
)

# Session factory — creates AsyncSession instances
AsyncSessionFactory = async_sessionmaker(
    engine,
    expire_on_commit=False,       # Keep attributes accessible after commit
    class_=AsyncSession,
)

async def get_db() -> AsyncSession:
    """Dependency: yield a DB session per request."""
    async with AsyncSessionFactory() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

Declarative ORM Models

app/db/base.py

from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
    pass

app/db/models/user.py

from datetime import datetime
from sqlalchemy import String, Boolean, DateTime, func
from sqlalchemy.orm import Mapped, mapped_column
from app.db.base import Base

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True, index=True)
    username: Mapped[str] = mapped_column(String(50), unique=True, nullable=False)
    email: Mapped[str] = mapped_column(String(255), unique=True, nullable=False, index=True)
    hashed_password: Mapped[str] = mapped_column(String(255), nullable=False)
    is_active: Mapped[bool] = mapped_column(Boolean, default=True, nullable=False)
    role: Mapped[str] = mapped_column(String(20), default="user", nullable=False)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now()
    )
    updated_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True), onupdate=func.now()
    )

SQLAlchemy 2 style uses Mapped[T] type annotations for full IDE support and type safety.

Relationships

app/db/models/post.py

from sqlalchemy import String, Text, ForeignKey
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.db.base import Base

class Post(Base):
    __tablename__ = "posts"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200), nullable=False)
    content: Mapped[str] = mapped_column(Text, nullable=False)
    author_id: Mapped[int] = mapped_column(ForeignKey("users.id"), nullable=False)

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

app/db/models/user.py (updated)

from sqlalchemy.orm import relationship

class User(Base):
    # ... existing fields ...
    posts: Mapped[list["Post"]] = relationship("Post", back_populates="author")

Creating Tables

For development and testing only — use Alembic for production migrations:

app/db/init_db.py

import asyncio
from app.db.base import Base
from app.db.session import engine
from app.db.models import user, post  # noqa: F401 — import to register models

async def create_tables() -> None:
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

if __name__ == "__main__":
    asyncio.run(create_tables())

python -m app.db.init_db

warning

Never use Base.metadata.create_all in production. Use Alembic migrations to manage schema changes safely.

Wiring into FastAPI

app/core/config.py

from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file=".env")
    DATABASE_URL: str = "postgresql+asyncpg://postgres:postgres@localhost:5432/app"
    DEBUG: bool = False

settings = Settings()

app/main.py

from contextlib import asynccontextmanager
from fastapi import FastAPI
from app.db.session import engine
from app.db.base import Base

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup: verify DB connection
    async with engine.connect() as conn:
        await conn.execute(text("SELECT 1"))
    yield
    # Shutdown: dispose engine
    await engine.dispose()

app = FastAPI(lifespan=lifespan)

SQLAlchemy 2 vs 1 Style

Feature	SQLAlchemy 1.x	SQLAlchemy 2.x
Column declaration	Column(Integer, ...)	mapped_column(...) with Mapped[int]
Query API	session.query(User).all()	select(User) + session.execute()
Async sessions	Extension package	Built-in with AsyncSession
Type safety	No IDE inference	Full Mapped[T] inference
Session factory	sessionmaker()	async_sessionmaker()

Common Pitfalls

Pitfall	Cause / Symptom	Fix
greenlet_spawn error	Calling sync SQLAlchemy in async context	Use AsyncSession with async driver
MissingGreenlet error	expire_on_commit=True + accessing attr after commit	Set expire_on_commit=False in async_sessionmaker
Session not rolled back on error	Missing except block in get_db	Wrap yield in try/except/rollback
Lazy loading MissingGreenlet error	Async session with lazy relationship loading	Use selectinload() or joinedload() in queries
asyncpg not found	Missing driver install	pip install asyncpg

Hands-On Practice

setup-db.sh

# Start PostgreSQL with Docker
docker run -d \
  --name fastapi-db \
  -e POSTGRES_USER=postgres \
  -e POSTGRES_PASSWORD=postgres \
  -e POSTGRES_DB=app \
  -p 5432:5432 \
  postgres:16-alpine

# Create .env
echo 'DATABASE_URL=postgresql+asyncpg://postgres:postgres@localhost:5432/app' > .env

# Install dependencies
pip install "sqlalchemy[asyncio]>=2.0" asyncpg pydantic-settings

# Initialize tables
python -m app.db.init_db

verify_connection.py

import asyncio
from sqlalchemy import text
from app.db.session import engine

async def check():
    async with engine.connect() as conn:
        result = await conn.execute(text("SELECT version()"))
        print(result.scalar())

asyncio.run(check())

What's Next

• CRUD Operations
• Migrations with Alembic
• Async Database Engines