SQLAlchemy Cheat Sheet

Overview

SQLAlchemy is the Python SQL toolkit and Object Relational Mapper that gives application developers the full power and flexibility of SQL. It provides a full suite of well-known enterprise-level persistence patterns, designed for efficient and high-performing database access. SQLAlchemy’s philosophy is that databases and SQL are powerful tools that shouldn’t be hidden behind abstraction walls.

Created by Mike Bayer in 2006, SQLAlchemy has become the de facto standard ORM for Python applications. It provides two distinct APIs: the Core layer for direct SQL expression construction, and the ORM layer for mapping Python classes to database tables. SQLAlchemy 2.0 introduced a modern, unified API with full type annotation support, making it more consistent and easier to use while maintaining backward compatibility.

Installation

Setup

# Base installation
pip install sqlalchemy

# With database drivers
pip install sqlalchemy psycopg2-binary    # PostgreSQL
pip install sqlalchemy pymysql            # MySQL
pip install sqlalchemy                     # SQLite (built-in)

# With async support
pip install sqlalchemy[asyncio] asyncpg   # Async PostgreSQL
pip install sqlalchemy[asyncio] aiosqlite # Async SQLite

# Type stubs
pip install sqlalchemy-stubs

Engine and Connection

Creating Engine

from sqlalchemy import create_engine

# SQLite
engine = create_engine("sqlite:///app.db", echo=True)

# PostgreSQL
engine = create_engine(
    "postgresql+psycopg2://user:password@localhost:5432/mydb",
    pool_size=10,
    max_overflow=20,
    pool_recycle=3600,
    echo=False,
)

# MySQL
engine = create_engine("mysql+pymysql://user:pass@localhost/mydb")

# Connection
with engine.connect() as conn:
    result = conn.execute(text("SELECT 1"))
    conn.commit()

Async Engine

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker

engine = create_async_engine(
    "postgresql+asyncpg://user:pass@localhost/mydb",
    echo=True,
)

async_session = async_sessionmaker(engine, expire_on_commit=False)

async with async_session() as session:
    result = await session.execute(select(User))
    users = result.scalars().all()

ORM Model Definition

Declarative Models (2.0 Style)

from datetime import datetime
from typing import Optional, List
from sqlalchemy import String, Integer, Boolean, ForeignKey, Text, DateTime, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))
    email: Mapped[str] = mapped_column(String(255), unique=True, index=True)
    age: Mapped[Optional[int]] = mapped_column(Integer, nullable=True)
    active: Mapped[bool] = mapped_column(Boolean, default=True)
    created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())
    updated_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now(), onupdate=func.now())

    # Relationships
    posts: Mapped[List["Post"]] = relationship(back_populates="author", cascade="all, delete-orphan")
    profile: Mapped[Optional["Profile"]] = relationship(back_populates="user", uselist=False)

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

class Post(Base):
    __tablename__ = "posts"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(255))
    body: Mapped[str] = mapped_column(Text)
    published: Mapped[bool] = mapped_column(Boolean, default=False)
    author_id: Mapped[int] = mapped_column(ForeignKey("users.id", ondelete="CASCADE"))

    author: Mapped["User"] = relationship(back_populates="posts")
    comments: Mapped[List["Comment"]] = relationship(back_populates="post", cascade="all, delete-orphan")
    created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())

class Comment(Base):
    __tablename__ = "comments"

    id: Mapped[int] = mapped_column(primary_key=True)
    content: Mapped[str] = mapped_column(Text)
    post_id: Mapped[int] = mapped_column(ForeignKey("posts.id", ondelete="CASCADE"))
    author_id: Mapped[int] = mapped_column(ForeignKey("users.id"))

    post: Mapped["Post"] = relationship(back_populates="comments")
    author: Mapped["User"] = relationship()

class Profile(Base):
    __tablename__ = "profiles"

    id: Mapped[int] = mapped_column(primary_key=True)
    bio: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), unique=True)

    user: Mapped["User"] = relationship(back_populates="profile")

Create Tables

# Create all tables
Base.metadata.create_all(engine)

# Drop all tables
Base.metadata.drop_all(engine)

Session and CRUD

Session Setup

from sqlalchemy.orm import Session, sessionmaker

# Session factory
SessionLocal = sessionmaker(bind=engine, expire_on_commit=False)

# Using session
with SessionLocal() as session:
    # operations
    session.commit()

# Context manager
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

Create

from sqlalchemy.orm import Session

with Session(engine) as session:
    # Single insert
    user = User(name="Alice", email="alice@example.com", age=30)
    session.add(user)
    session.commit()
    session.refresh(user)  # Get generated ID

    # Bulk insert
    users = [
        User(name="Bob", email="bob@example.com"),
        User(name="Charlie", email="charlie@example.com"),
    ]
    session.add_all(users)
    session.commit()

    # Create with relationships
    user = User(
        name="Dave",
        email="dave@example.com",
        posts=[
            Post(title="First Post", body="Content"),
            Post(title="Second Post", body="More content"),
        ],
        profile=Profile(bio="Hello!"),
    )
    session.add(user)
    session.commit()

Read

from sqlalchemy import select, and_, or_, func, desc

with Session(engine) as session:
    # Get by primary key
    user = session.get(User, 1)

    # Select all
    stmt = select(User)
    users = session.scalars(stmt).all()

    # Filtering
    stmt = select(User).where(User.active == True).where(User.age >= 18)
    users = session.scalars(stmt).all()

    # Complex conditions
    stmt = select(User).where(
        and_(
            User.active == True,
            or_(
                User.role == "admin",
                User.age >= 18,
            )
        )
    ).order_by(desc(User.created_at)).limit(10).offset(0)

    # Like / ilike
    stmt = select(User).where(User.name.ilike("%alice%"))

    # In clause
    stmt = select(User).where(User.id.in_([1, 2, 3]))

    # Join
    stmt = (
        select(User, Post)
        .join(Post, User.id == Post.author_id)
        .where(Post.published == True)
    )
    results = session.execute(stmt).all()

    # Eager loading
    from sqlalchemy.orm import joinedload, selectinload
    stmt = select(User).options(
        joinedload(User.profile),
        selectinload(User.posts),
    ).where(User.id == 1)
    user = session.scalars(stmt).unique().one()

    # Aggregation
    stmt = select(func.count(User.id), func.avg(User.age)).where(User.active == True)
    count, avg_age = session.execute(stmt).one()

    # Group by
    stmt = (
        select(User.role, func.count(User.id))
        .group_by(User.role)
    )
    results = session.execute(stmt).all()

    # Subquery
    subq = (
        select(Post.author_id, func.count(Post.id).label("post_count"))
        .group_by(Post.author_id)
        .subquery()
    )
    stmt = select(User.name, subq.c.post_count).join(subq, User.id == subq.c.author_id)

Update

with Session(engine) as session:
    # Update by object
    user = session.get(User, 1)
    user.name = "Alice Smith"
    session.commit()

    # Bulk update
    from sqlalchemy import update
    stmt = (
        update(User)
        .where(User.active == False)
        .values(role="inactive")
    )
    session.execute(stmt)
    session.commit()

Delete

with Session(engine) as session:
    # Delete by object
    user = session.get(User, 1)
    session.delete(user)
    session.commit()

    # Bulk delete
    from sqlalchemy import delete
    stmt = delete(User).where(User.active == False)
    session.execute(stmt)
    session.commit()

Configuration

Alembic Migrations

# Install Alembic
pip install alembic

# Initialize
alembic init alembic

# Generate migration
alembic revision --autogenerate -m "create users table"

# Apply migrations
alembic upgrade head

# Rollback
alembic downgrade -1

# alembic/env.py
from app.models import Base
target_metadata = Base.metadata

Advanced Usage

Events

from sqlalchemy import event

@event.listens_for(User, "before_insert")
def set_defaults(mapper, connection, target):
    if not target.role:
        target.role = "user"

@event.listens_for(Session, "after_flush")
def after_flush(session, flush_context):
    for obj in session.new:
        print(f"Created: {obj}")

Hybrid Properties

from sqlalchemy.ext.hybrid import hybrid_property

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    first_name: Mapped[str] = mapped_column(String(50))
    last_name: Mapped[str] = mapped_column(String(50))

    @hybrid_property
    def full_name(self) -> str:
        return f"{self.first_name} {self.last_name}"

    @full_name.expression
    @classmethod
    def full_name(cls):
        return cls.first_name + " " + cls.last_name

# Works in queries
stmt = select(User).where(User.full_name == "Alice Smith")

Raw SQL

from sqlalchemy import text

with Session(engine) as session:
    result = session.execute(
        text("SELECT * FROM users WHERE age > :age"),
        {"age": 18}
    )
    rows = result.fetchall()

Troubleshooting

Problem	Solution
DetachedInstanceError	Use expire_on_commit=False or reload object in new session
N+1 queries	Use joinedload() or selectinload() for eager loading
IntegrityError on insert	Check unique constraints; handle with try/except
Session not committing	Call session.commit() explicitly; check for exceptions
Stale data	Use session.refresh(obj) or session.expire(obj)
Slow queries	Enable echo=True on engine; add indexes to models
Migration autogenerate misses changes	Ensure target_metadata points to your Base.metadata
Connection pool exhaustion	Increase pool_size; ensure sessions are properly closed