SQLAlchemy Setup — Engine, Session and Declarative Models

SQLAlchemy is the most widely used Python ORM — it maps Python classes to database tables, generates SQL from Python expressions, and manages database sessions (connections with transaction state). SQLAlchemy 2.x introduced a modern, fully typed API using Python’s type hints directly in model definitions, making ORM models self-documenting and IDE-friendly. In a FastAPI application, SQLAlchemy provides the database access layer: models define the schema, sessions manage transactions, and the get_db dependency injects a session into each route handler.

Installation and Engine Setup #

pip install sqlalchemy psycopg2-binary
# For async: pip install sqlalchemy[asyncio] asyncpg

# app/database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, DeclarativeBase
from app.config import settings

# ── Engine — the connection pool ──────────────────────────────────────────────
engine = create_engine(
    str(settings.database_url),
    pool_size     = settings.db_pool_size,
    max_overflow  = 10,
    pool_pre_ping = True,     # test connection before use
    echo          = settings.db_echo_sql,  # log SQL in development
)

# ── Session factory ───────────────────────────────────────────────────────────
SessionLocal = sessionmaker(
    autocommit = False,   # transactions are explicit
    autoflush  = False,   # don't auto-flush before queries
    bind       = engine,
)

# ── Declarative base ──────────────────────────────────────────────────────────
class Base(DeclarativeBase):
    pass
# All ORM models inherit from Base
# Base.metadata contains all table definitions (used by Alembic)

The FastAPI Database Dependency #

# app/dependencies.py
from typing import Generator
from sqlalchemy.orm import Session
from app.database import SessionLocal

def get_db() -> Generator[Session, None, None]:
    """
    FastAPI dependency that provides a database session per request.
    Commits on success, rolls back on exception, always closes the session.
    """
    db = SessionLocal()
    try:
        yield db
        db.commit()
    except Exception:
        db.rollback()
        raise
    finally:
        db.close()   # returns connection to pool

# Usage in route handler:
# from fastapi import Depends
# from sqlalchemy.orm import Session
# from app.dependencies import get_db
#
# @router.get("/posts")
# def list_posts(db: Session = Depends(get_db)):
#     return db.query(Post).all()

ORM Model Definitions #

# app/models/user.py
from sqlalchemy import String, Boolean, Text
from sqlalchemy.orm import Mapped, mapped_column, relationship
from sqlalchemy.sql import func
from datetime import datetime
from app.database import Base

class User(Base):
    __tablename__ = "users"

    # ── Columns ───────────────────────────────────────────────────────────────
    id:            Mapped[int]      = mapped_column(primary_key=True, index=True)
    email:         Mapped[str]      = mapped_column(String(320), unique=True, index=True)
    name:          Mapped[str]      = mapped_column(String(100))
    password_hash: Mapped[str]      = mapped_column(Text)
    role:          Mapped[str]      = mapped_column(String(20), default="user")
    is_active:     Mapped[bool]     = mapped_column(Boolean, default=True)
    created_at:    Mapped[datetime] = mapped_column(server_default=func.now())
    updated_at:    Mapped[datetime] = mapped_column(
                       server_default=func.now(), onupdate=func.now()
                   )

    # ── Relationships (defined in next lesson) ────────────────────────────────
    posts:    Mapped[list["Post"]]    = relationship(back_populates="author")
    comments: Mapped[list["Comment"]] = relationship(back_populates="author")

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

SQLAlchemy 2.x Typed Column Syntax #

from sqlalchemy import String, Text, Integer, Boolean, ForeignKey, TIMESTAMP
from sqlalchemy.dialects.postgresql import UUID, JSONB
from sqlalchemy.orm import Mapped, mapped_column
from datetime import datetime
from uuid import UUID as PyUUID

# Mapped[type] declares the Python type and nullability:
# Mapped[int]        → non-nullable integer column
# Mapped[int | None] → nullable integer column (Optional[int])
# Mapped[str]        → non-nullable text column

class Post(Base):
    __tablename__ = "posts"

    id:         Mapped[int]           = mapped_column(primary_key=True)
    public_id:  Mapped[PyUUID]        = mapped_column(UUID(as_uuid=True), unique=True)
    title:      Mapped[str]           = mapped_column(String(200))
    body:       Mapped[str]           = mapped_column(Text)
    slug:       Mapped[str]           = mapped_column(String(200), unique=True)
    status:     Mapped[str]           = mapped_column(String(20), default="draft")
    view_count: Mapped[int]           = mapped_column(Integer, default=0)
    metadata_:  Mapped[dict | None]   = mapped_column("metadata", JSONB, nullable=True)
    author_id:  Mapped[int]           = mapped_column(ForeignKey("users.id"))
    created_at: Mapped[datetime]      = mapped_column(server_default=func.now())
    updated_at: Mapped[datetime]      = mapped_column(
                    server_default=func.now(), onupdate=func.now()
                )

Common Mistakes #

Mistake 1 — Creating engine inside the dependency (no connection pooling) #

❌ Wrong — new engine per request:

def get_db():
    engine = create_engine(DATABASE_URL)   # new pool every request!
    db = Session(engine)
    yield db

✅ Correct — module-level engine, session factory per request:

engine = create_engine(DATABASE_URL)   # module-level — created once ✓
SessionLocal = sessionmaker(bind=engine)
def get_db():
    db = SessionLocal()
    yield db   # reuses connections from pool ✓

Mistake 2 — Not calling db.close() after the request #

❌ Wrong — connection never returned to pool:

def get_db():
    db = SessionLocal()
    yield db
    # session never closed — connection pool exhausted after N requests!

✅ Correct — always close in finally:

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()   # ✓ returns connection to pool

Mistake 3 — Using Mapped without server_default for auto-timestamp columns #

❌ Wrong — Python sets the timestamp (uses application timezone, not DB timezone):

from datetime import datetime
created_at: Mapped[datetime] = mapped_column(default=datetime.utcnow)

✅ Correct — let the database set it with server_default:

from sqlalchemy.sql import func
created_at: Mapped[datetime] = mapped_column(server_default=func.now())   # ✓

Quick Reference #