Python SQLAlchemy

A powerful Python SQL toolkit and Object-Relational Mapper (ORM) for database access and manipulation.

Overview

SQLAlchemy provides a full suite of enterprise-level persistence patterns for efficient database access. It consists of two main components: the Core (SQL expression language) and the ORM (object-relational mapping). SQLAlchemy supports PostgreSQL, MySQL, SQLite, Oracle, and other databases through a unified API.

⤢ zoom

Engine and Session Setup

The engine is the starting point for any SQLAlchemy application, managing database connections and serving as the source of connectivity.

Key Concepts

• Engine: Factory for database connections and connection pool manager
• Session: Manages persistence operations and transaction scope
• Connection pool: Reusable database connections for performance
• sessionmaker: Factory for creating session instances with consistent configuration

Common Patterns

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, scoped_session

# Basic engine creation
engine = create_engine("postgresql://user:password@localhost:5432/dbname")

# SQLite (file-based)
engine = create_engine("sqlite:///app.db")

# SQLite (in-memory)
engine = create_engine("sqlite:///:memory:")

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

# Engine with connection pool configuration
engine = create_engine(
    "postgresql://user:password@localhost:5432/dbname",
    pool_size=10,           # Number of connections in pool
    max_overflow=20,        # Extra connections beyond pool_size
    pool_pre_ping=True,     # Test connections before use
    pool_recycle=3600,      # Recycle connections after 1 hour
    echo=False              # Set True for SQL logging
)

# Create session factory
SessionLocal = sessionmaker(
    bind=engine,
    autocommit=False,
    autoflush=False
)

# Create session instance
session = SessionLocal()

# Thread-safe scoped session (for web applications)
Session = scoped_session(sessionmaker(bind=engine))

Examples

FastAPI/Flask dependency pattern:

from sqlalchemy import create_engine, select
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "postgresql://user:password@localhost:5432/dbname"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)

# Dependency function for FastAPI
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# Usage in FastAPI
from fastapi import Depends

@app.get("/users")
def get_users(db: Session = Depends(get_db)):
    return db.scalars(select(User)).all()

Async engine setup (SQLAlchemy 2.0+):

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker

# Async engine
async_engine = create_async_engine(
    "postgresql+asyncpg://user:password@localhost:5432/dbname",
    echo=True
)

# Async session factory
AsyncSessionLocal = async_sessionmaker(
    bind=async_engine,
    expire_on_commit=False
)

# Usage
async def get_users():
    async with AsyncSessionLocal() as session:
        result = await session.execute(select(User))
        return result.scalars().all()

Model Definition

Models are Python classes that map to database tables using the declarative system.

Key Concepts

• declarative_base: Base class for all ORM models
• Column: Defines table columns with types and constraints
• Primary key: Unique identifier for each row
• Nullable: Whether column allows NULL values
• Default/server_default: Default values for columns

Common Patterns

from sqlalchemy import (
    Column, Integer, String, Float, Boolean, DateTime,
    Text, Date, Time, Numeric, LargeBinary, JSON,
    ForeignKey, UniqueConstraint, Index, CheckConstraint
)
from sqlalchemy.orm import declarative_base
from sqlalchemy.sql import func
from datetime import datetime, timezone

Base = declarative_base()

class User(Base):
    __tablename__ = "users"

    # Primary key
    id = Column(Integer, primary_key=True, index=True)

    # String types
    username = Column(String(50), unique=True, nullable=False)
    email = Column(String(255), unique=True, nullable=False, index=True)
    bio = Column(Text, nullable=True)

    # Numeric types
    age = Column(Integer, nullable=True)
    balance = Column(Numeric(10, 2), default=0.00)

    # Boolean
    is_active = Column(Boolean, default=True, nullable=False)

    # DateTime with defaults (datetime.utcnow is deprecated since Python 3.12)
    created_at = Column(DateTime, default=lambda: datetime.now(timezone.utc))
    updated_at = Column(
        DateTime,
        default=lambda: datetime.now(timezone.utc),
        onupdate=lambda: datetime.now(timezone.utc)
    )

    # Server-side default (database handles it)
    registered_at = Column(
        DateTime,
        server_default=func.now()
    )

    # JSON column (PostgreSQL, MySQL 5.7+, SQLite 3.9+)
    # Use a callable, not a literal {} — a literal would be shared between rows
    preferences = Column(JSON, default=dict)

    # Table-level constraints
    __table_args__ = (
        UniqueConstraint("email", name="uq_user_email"),
        Index("ix_user_username_email", "username", "email"),
        CheckConstraint("age >= 0", name="ck_user_age_positive"),
    )

    def __repr__(self):
        return f"<User(id={self.id}, username='{self.username}')>"


class Product(Base):
    __tablename__ = "products"

    id = Column(Integer, primary_key=True)
    name = Column(String(100), nullable=False)
    price = Column(Float, nullable=False)
    sku = Column(String(50), unique=True)

    # Composite primary key example
    # id = Column(Integer, primary_key=True)
    # version = Column(Integer, primary_key=True)


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

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

Examples

Using Mapped and mapped_column (SQLAlchemy 2.0+ style):

from sqlalchemy.orm import Mapped, mapped_column, DeclarativeBase
from typing import Optional

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    username: Mapped[str] = mapped_column(String(50), unique=True)
    email: Mapped[str] = mapped_column(String(255))
    bio: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
    is_active: Mapped[bool] = mapped_column(default=True)

Mixins for common columns:

from sqlalchemy import Column, Integer, DateTime
from sqlalchemy.sql import func

class TimestampMixin:
    created_at = Column(DateTime, server_default=func.now())
    updated_at = Column(DateTime, onupdate=func.now())

class IdMixin:
    id = Column(Integer, primary_key=True, index=True)

class User(IdMixin, TimestampMixin, Base):
    __tablename__ = "users"
    username = Column(String(50), nullable=False)

CRUD Operations

Basic Create, Read, Update, and Delete operations using the Session API.

⤢ zoom

Key Concepts

• add(): Stage object for insertion
• commit(): Persist all staged changes to database
• flush(): Send pending changes to database without committing
• rollback(): Revert uncommitted changes
• refresh(): Reload object state from database

Common Patterns

from sqlalchemy import select, update, delete
from sqlalchemy.orm import Session

# CREATE - Insert new records
def create_user(db: Session, username: str, email: str):
    user = User(username=username, email=email)
    db.add(user)
    db.commit()
    db.refresh(user)  # Load database-generated values
    return user

# Bulk insert
def create_users_bulk(db: Session, users_data: list):
    users = [User(**data) for data in users_data]
    db.add_all(users)
    db.commit()
    return users

# READ - Query records
def get_user(db: Session, user_id: int):
    return db.get(User, user_id)  # Primary key lookup

def get_user_by_email(db: Session, email: str):
    return db.scalar(select(User).where(User.email == email))

def get_users(db: Session, skip: int = 0, limit: int = 100):
    return db.scalars(select(User).offset(skip).limit(limit)).all()

def get_active_users(db: Session):
    return db.scalars(select(User).where(User.is_active == True)).all()

# UPDATE - Modify existing records
def update_user(db: Session, user_id: int, **kwargs):
    user = db.get(User, user_id)
    if user:
        for key, value in kwargs.items():
            setattr(user, key, value)
        db.commit()
        db.refresh(user)
    return user

# Bulk update
def deactivate_old_users(db: Session, days: int = 365):
    from datetime import datetime, timedelta, timezone
    cutoff = datetime.now(timezone.utc) - timedelta(days=days)

    result = db.execute(
        update(User)
        .where(User.last_login < cutoff)
        .values(is_active=False)
    )
    db.commit()
    return result.rowcount  # Number of rows updated

# DELETE - Remove records
def delete_user(db: Session, user_id: int):
    user = db.get(User, user_id)
    if user:
        db.delete(user)
        db.commit()
        return True
    return False

# Bulk delete
def delete_inactive_users(db: Session):
    result = db.execute(delete(User).where(User.is_active == False))
    db.commit()
    return result.rowcount

Legacy note: the 1.x-style Query API (db.query(User).filter(...).first()) still works on SQLAlchemy 2.x as "2.0-legacy" and remains common in older codebases, but select() + session.execute()/scalars() is the current idiom and the only style covered here.

Examples

Using session.get() for primary key lookup:

# Checks the identity map first; avoids a query if already loaded
user = db.get(User, user_id)

# With composite primary key
item = db.get(OrderItem, (order_id, product_id))

Merge for upsert-like behaviour:

def upsert_user(db: Session, user_data: dict):
    user = User(**user_data)
    merged = db.merge(user)  # Updates if exists, inserts if not
    db.commit()
    return merged

Context manager pattern:

from contextlib import contextmanager

@contextmanager
def get_session():
    session = SessionLocal()
    try:
        yield session
        session.commit()
    except Exception:
        session.rollback()
        raise
    finally:
        session.close()

# Usage
with get_session() as db:
    user = User(username="john", email="john@example.com")
    db.add(user)
    # Auto-commits on successful exit

Relationships

Define relationships between models using foreign keys and the relationship() function.

⤢ zoom

Key Concepts

• ForeignKey: Database-level constraint linking tables
• relationship(): ORM-level relationship for object access
• backref/back_populates: Bidirectional relationship access
• Lazy loading: Load related objects on demand
• Eager loading: Load related objects upfront

Common Patterns

from sqlalchemy import Column, Integer, String, ForeignKey, Table
from sqlalchemy.orm import relationship, backref

# One-to-Many relationship
class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True)
    username = Column(String(50), nullable=False)

    # One user has many posts
    posts = relationship("Post", back_populates="author")

    # Alternative using backref (defines both sides)
    # posts = relationship("Post", backref="author")

class Post(Base):
    __tablename__ = "posts"

    id = Column(Integer, primary_key=True)
    title = Column(String(200), nullable=False)
    user_id = Column(Integer, ForeignKey("users.id"), nullable=False)

    # Many posts belong to one user
    author = relationship("User", back_populates="posts")


# One-to-One relationship
class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True)
    username = Column(String(50))

    # uselist=False for one-to-one
    profile = relationship(
        "Profile",
        back_populates="user",
        uselist=False
    )

class Profile(Base):
    __tablename__ = "profiles"

    id = Column(Integer, primary_key=True)
    user_id = Column(Integer, ForeignKey("users.id"), unique=True)
    bio = Column(Text)

    user = relationship("User", back_populates="profile")


# Many-to-Many relationship
post_tags = Table(
    "post_tags",
    Base.metadata,
    Column("post_id", Integer, ForeignKey("posts.id"), primary_key=True),
    Column("tag_id", Integer, ForeignKey("tags.id"), primary_key=True)
)

class Post(Base):
    __tablename__ = "posts"

    id = Column(Integer, primary_key=True)
    title = Column(String(200))

    tags = relationship(
        "Tag",
        secondary=post_tags,
        back_populates="posts"
    )

class Tag(Base):
    __tablename__ = "tags"

    id = Column(Integer, primary_key=True)
    name = Column(String(50), unique=True)

    posts = relationship(
        "Post",
        secondary=post_tags,
        back_populates="tags"
    )

Examples

Cascade delete:

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True)
    username = Column(String(50))

    # Delete all posts when user is deleted
    posts = relationship(
        "Post",
        back_populates="author",
        cascade="all, delete-orphan"
    )

Lazy loading options:

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True)

    # Different lazy loading strategies
    posts = relationship("Post", lazy="select")         # Default, load on access
    comments = relationship("Comment", lazy="selectin")  # Batched SELECT ... IN
    addresses = relationship("Address", lazy="joined")  # JOIN on initial load
    orders = relationship("Order", lazy="dynamic")      # Returns a query-like object

# Note: lazy="subquery" is superseded by lazy="selectin" in 2.x guidance —
# selectin emits a simpler, usually faster SELECT ... WHERE id IN (...)

Self-referential relationship (tree structure):

class Category(Base):
    __tablename__ = "categories"

    id = Column(Integer, primary_key=True)
    name = Column(String(100))
    parent_id = Column(Integer, ForeignKey("categories.id"))

    # Parent/child relationship
    children = relationship(
        "Category",
        backref=backref("parent", remote_side=[id])
    )

Working with relationships:

# Create with relationship
user = User(username="john")
post = Post(title="First Post", author=user)
db.add(post)
db.commit()

# Access related objects
user = db.scalars(select(User)).first()
for post in user.posts:
    print(post.title)

# Add to relationship
new_post = Post(title="Second Post")
user.posts.append(new_post)
db.commit()

# Many-to-many
post = Post(title="Tagged Post")
tag = Tag(name="python")
post.tags.append(tag)
db.add(post)
db.commit()

Query Building

Advanced query construction using select() and the SQL expression language.

Key Concepts

• select(): Builds a SELECT statement; executed via session.execute() or session.scalars()
• where(): SQL WHERE clause with expressions (filter_by() for keyword shorthand)
• join(): JOIN operations between tables
• order_by(): Sort results
• group_by(): Aggregate results

Common Patterns

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

# Basic filtering
users = db.scalars(select(User).where(User.is_active == True)).all()
users = db.scalars(select(User).filter_by(is_active=True)).all()  # Keyword shorthand

# Multiple conditions (AND)
users = db.scalars(
    select(User).where(
        User.is_active == True,
        User.age >= 18
    )
).all()

# Explicit AND/OR
users = db.scalars(
    select(User).where(
        and_(
            User.is_active == True,
            or_(
                User.role == "admin",
                User.role == "moderator"
            )
        )
    )
).all()

# NOT condition
inactive = db.scalars(select(User).where(not_(User.is_active))).all()

# LIKE queries
users = db.scalars(select(User).where(User.email.like("%@gmail.com"))).all()
users = db.scalars(select(User).where(User.username.ilike("%john%"))).all()  # Case-insensitive

# IN clause
users = db.scalars(select(User).where(User.id.in_([1, 2, 3]))).all()
users = db.scalars(select(User).where(~User.id.in_([1, 2, 3]))).all()  # NOT IN

# NULL checks
users = db.scalars(select(User).where(User.bio.is_(None))).all()
users = db.scalars(select(User).where(User.bio.is_not(None))).all()

# BETWEEN
users = db.scalars(select(User).where(User.age.between(18, 65))).all()

# Ordering
users = db.scalars(select(User).order_by(User.created_at.desc())).all()
users = db.scalars(select(User).order_by(desc(User.created_at), asc(User.username))).all()

# Limiting and offset
users = db.scalars(select(User).order_by(User.id).limit(10).offset(20)).all()

# First result
user = db.scalars(select(User).where(User.username == "john")).first()

# One result (raises if not exactly one)
user = db.scalars(select(User).where(User.username == "john")).one()

# One or none
user = db.scalars(select(User).where(User.username == "john")).one_or_none()

# Count
total = db.scalar(
    select(func.count()).select_from(User).where(User.is_active == True)
)

# Exists check
exists = db.scalar(select(User.id).where(User.email == email).limit(1)) is not None

Examples

JOINs:

# Implicit join (using relationship)
posts = db.scalars(
    select(Post).join(Post.author).where(User.is_active == True)
).all()

# Explicit join
posts = db.scalars(
    select(Post).join(User, Post.user_id == User.id).where(User.is_active == True)
).all()

# Left outer join
posts = db.scalars(select(Post).outerjoin(User)).all()

# Multiple joins
results = db.scalars(
    select(Post).join(User).join(Category).where(Category.name == "Technology")
).all()

# Select from multiple tables (rows of tuples, so use execute, not scalars)
results = db.execute(
    select(User, Post).join(Post).where(Post.created_at > some_date)
).all()

for user, post in results:
    print(f"{user.username}: {post.title}")

Aggregations:

from sqlalchemy import select, func

# Count
total_users = db.scalar(select(func.count(User.id)))

# Sum
total_balance = db.scalar(select(func.sum(User.balance)))

# Average
avg_age = db.scalar(select(func.avg(User.age)))

# Group by
posts_per_user = db.execute(
    select(User.username, func.count(Post.id).label("post_count"))
    .join(Post)
    .group_by(User.id)
).all()

# Having clause
active_authors = db.execute(
    select(User.username, func.count(Post.id).label("post_count"))
    .join(Post)
    .group_by(User.id)
    .having(func.count(Post.id) > 5)
).all()

Subqueries:

# Subquery for users with most recent post
latest_post = (
    select(func.max(Post.created_at))
    .where(Post.user_id == User.id)
    .correlate(User)
    .scalar_subquery()
)

users_with_latest = db.execute(
    select(User.username, latest_post.label("latest_post_date"))
).all()

# EXISTS subquery
from sqlalchemy import exists

has_posts = db.scalars(
    select(User).where(exists().where(Post.user_id == User.id))
).all()

Eager loading to avoid N+1:

from sqlalchemy.orm import joinedload, selectinload

# selectinload - separate SELECT ... IN; preferred for collections
users = db.scalars(
    select(User).options(selectinload(User.posts))
).all()

# joinedload - single query with JOIN; best for many-to-one / one-to-one
# (.unique() is required when joinedload targets a collection)
users = db.scalars(
    select(User).options(joinedload(User.posts))
).unique().all()

# Nested eager loading
users = db.scalars(
    select(User).options(selectinload(User.posts).selectinload(Post.tags))
).all()

# Multiple relationships
users = db.scalars(
    select(User).options(
        selectinload(User.posts),
        joinedload(User.profile)
    )
).all()

Transactions and Session Management

Control transaction boundaries and session lifecycle for data integrity.

⤢ zoom

Key Concepts

• Transaction: Unit of work that succeeds or fails entirely
• commit(): Persist all changes and end transaction
• rollback(): Discard all changes and end transaction
• begin_nested(): Create savepoint for partial rollbacks
• Object states: Transient, pending, persistent, detached, deleted

Common Patterns

from sqlalchemy.orm import Session
from sqlalchemy.exc import IntegrityError, SQLAlchemyError

# Basic transaction
def transfer_funds(db: Session, from_id: int, to_id: int, amount: float):
    try:
        from_account = db.get(Account, from_id)
        to_account = db.get(Account, to_id)

        if from_account.balance < amount:
            raise ValueError("Insufficient funds")

        from_account.balance -= amount
        to_account.balance += amount

        db.commit()
        return True
    except Exception as e:
        db.rollback()
        raise

# Savepoints for partial rollbacks
def process_orders(db: Session, orders: list):
    for order in orders:
        savepoint = db.begin_nested()
        try:
            process_order(db, order)
            savepoint.commit()
        except Exception as e:
            savepoint.rollback()
            log_failed_order(order, e)

    db.commit()  # Commit all successful orders

# Context manager with auto-rollback
from contextlib import contextmanager

@contextmanager
def transaction(session_factory):
    session = session_factory()
    try:
        yield session
        session.commit()
    except Exception:
        session.rollback()
        raise
    finally:
        session.close()

# Usage
with transaction(SessionLocal) as db:
    user = User(username="john")
    db.add(user)
    # Auto-commits or rolls back

Examples

Handling concurrent updates with locking:

from sqlalchemy.orm import Session

# Pessimistic locking (SELECT ... FOR UPDATE)
def update_inventory(db: Session, product_id: int, quantity: int):
    product = db.scalars(
        select(Product).where(Product.id == product_id).with_for_update()
    ).first()

    if product.stock < quantity:
        raise ValueError("Insufficient stock")

    product.stock -= quantity
    db.commit()

# Optimistic locking with version column
class Product(Base):
    __tablename__ = "products"

    id = Column(Integer, primary_key=True)
    name = Column(String(100))
    stock = Column(Integer)
    version = Column(Integer, nullable=False, default=1)

    __mapper_args__ = {
        "version_id_col": version
    }

Bulk operations for performance:

from sqlalchemy.dialects.postgresql import insert

# Bulk insert with INSERT ... ON CONFLICT (PostgreSQL)
def upsert_products(db: Session, products: list):
    stmt = insert(Product).values(products)
    stmt = stmt.on_conflict_do_update(
        index_elements=["sku"],
        set_={"price": stmt.excluded.price}
    )
    db.execute(stmt)
    db.commit()

# Bulk update
from sqlalchemy import update

db.execute(
    update(User)
    .where(User.is_active == False)
    .values(is_active=True),
    execution_options={"synchronize_session": "fetch"},
)
db.commit()

Session events:

from sqlalchemy import event

@event.listens_for(Session, "before_commit")
def before_commit(session):
    for obj in session.new:
        if hasattr(obj, "created_at"):
            obj.created_at = datetime.now(timezone.utc)

@event.listens_for(Session, "after_rollback")
def after_rollback(session):
    logging.warning("Transaction rolled back")

Common Patterns and Best Practices

Repository Pattern

from typing import TypeVar, Generic, Type
from sqlalchemy import select
from sqlalchemy.orm import Session

T = TypeVar("T")

class BaseRepository(Generic[T]):
    def __init__(self, db: Session, model: Type[T]):
        self.db = db
        self.model = model

    def get(self, id: int) -> T | None:
        return self.db.get(self.model, id)

    def get_all(self, skip: int = 0, limit: int = 100) -> list[T]:
        stmt = select(self.model).offset(skip).limit(limit)
        return list(self.db.scalars(stmt))

    def create(self, **kwargs) -> T:
        obj = self.model(**kwargs)
        self.db.add(obj)
        self.db.commit()
        self.db.refresh(obj)
        return obj

    def update(self, id: int, **kwargs) -> T | None:
        obj = self.get(id)
        if obj:
            for key, value in kwargs.items():
                setattr(obj, key, value)
            self.db.commit()
            self.db.refresh(obj)
        return obj

    def delete(self, id: int) -> bool:
        obj = self.get(id)
        if obj:
            self.db.delete(obj)
            self.db.commit()
            return True
        return False

class UserRepository(BaseRepository[User]):
    def __init__(self, db: Session):
        super().__init__(db, User)

    def get_by_email(self, email: str) -> User | None:
        return self.db.scalar(select(User).where(User.email == email))

    def get_active_users(self) -> list[User]:
        return list(self.db.scalars(select(User).where(User.is_active == True)))

Soft Delete Pattern

from datetime import datetime, timezone
from sqlalchemy import Column, DateTime, Boolean, event, select
from sqlalchemy.orm import Session, with_loader_criteria

class SoftDeleteMixin:
    deleted_at = Column(DateTime, nullable=True)
    is_deleted = Column(Boolean, default=False, nullable=False)

    def soft_delete(self):
        self.is_deleted = True
        self.deleted_at = datetime.now(timezone.utc)

# Filter soft-deleted rows out of every ORM SELECT automatically
@event.listens_for(Session, "do_orm_execute")
def filter_soft_deleted(execute_state):
    if (
        execute_state.is_select
        and not execute_state.is_column_load
        and not execute_state.is_relationship_load
        and not execute_state.execution_options.get("include_deleted", False)
    ):
        execute_state.statement = execute_state.statement.options(
            with_loader_criteria(
                SoftDeleteMixin,
                lambda cls: cls.is_deleted == False,
                include_aliases=True,
            )
        )

# Usage
class User(SoftDeleteMixin, Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    username = Column(String(50))

# Normal queries exclude soft-deleted rows automatically
users = db.scalars(select(User)).all()

# Opt back in per query when needed
all_users = db.scalars(
    select(User).execution_options(include_deleted=True)
).all()

Pagination Helper

from dataclasses import dataclass
from typing import TypeVar, Generic
from sqlalchemy import select, func
from sqlalchemy.orm import Session

T = TypeVar("T")

@dataclass
class Page(Generic[T]):
    items: list[T]
    total: int
    page: int
    size: int
    pages: int

def paginate(db: Session, stmt, page: int = 1, size: int = 20) -> Page:
    total = db.scalar(select(func.count()).select_from(stmt.subquery()))
    pages = (total + size - 1) // size
    items = db.scalars(stmt.offset((page - 1) * size).limit(size)).all()

    return Page(
        items=items,
        total=total,
        page=page,
        size=size,
        pages=pages
    )

# Usage
stmt = select(User).where(User.is_active == True)
result = paginate(db, stmt, page=2, size=10)

Testing with SQLAlchemy

import pytest
from sqlalchemy import create_engine, select
from sqlalchemy.orm import sessionmaker

@pytest.fixture
def engine():
    return create_engine("sqlite:///:memory:")

@pytest.fixture
def tables(engine):
    Base.metadata.create_all(engine)
    yield
    Base.metadata.drop_all(engine)

@pytest.fixture
def db_session(engine, tables):
    Session = sessionmaker(bind=engine)
    session = Session()
    yield session
    session.rollback()
    session.close()

def test_create_user(db_session):
    user = User(username="test", email="test@example.com")
    db_session.add(user)
    db_session.commit()

    result = db_session.scalars(
        select(User).filter_by(username="test")
    ).first()
    assert result is not None
    assert result.email == "test@example.com"

Quick Reference

Task	Code
Create engine	create_engine("postgresql://user:pass@host/db")
Create session	SessionLocal = sessionmaker(bind=engine)
Create tables	Base.metadata.create_all(engine)
Add object	db.add(obj)
Commit changes	db.commit()
Rollback	db.rollback()
Query all	db.scalars(select(Model)).all()
Filter	db.scalars(select(Model).where(Model.col == val)).all()
Filter by kwargs	db.scalars(select(Model).filter_by(col=val)).all()
First result	db.scalars(select(Model)).first()
Single scalar	db.scalar(select(func.count(Model.id)))
Get by PK	db.get(Model, id)
Order by	select(Model).order_by(Model.col.desc())
Join	select(Model).join(Related)
Count	db.scalar(select(func.count()).select_from(Model))
Bulk update	db.execute(update(Model).where(...).values(...))
Bulk delete	db.execute(delete(Model).where(...))
Delete object	db.delete(obj)
Eager load	db.scalars(select(Model).options(selectinload(Model.rel))).all()

Common Imports

from sqlalchemy import (
    create_engine, select, update, delete,
    Column, Integer, String, Float, Boolean,
    DateTime, Text, ForeignKey, Table, and_, or_, not_,
    desc, asc, func
)
from sqlalchemy.orm import (
    declarative_base, sessionmaker, relationship, backref,
    joinedload, selectinload, Session
)
from sqlalchemy.exc import IntegrityError, SQLAlchemyError

Column Types

Python Type	SQLAlchemy Column
int	Integer
str	String(length), Text
float	Float, Numeric(precision, scale)
bool	Boolean
datetime	DateTime, Date, Time
bytes	LargeBinary
dict	JSON
uuid	UUID (PostgreSQL)
enum	Enum

Common Issues and Solutions

Issue	Solution
DetachedInstanceError	Access attributes before closing session, or use db.refresh(obj)
InvalidRequestError: already attached	Don't add same object to multiple sessions; use merge() instead
N+1 query problem	Use eager loading: joinedload(), selectinload()
IntegrityError on unique constraint	Catch exception and handle duplicate; use ON CONFLICT for upserts
Session not thread-safe	Use scoped_session() or create session per request
Stale data after commit	Call db.refresh(obj) to reload from database
Connection pool exhausted	Increase pool_size, ensure sessions are closed, use context managers
Circular import with models	Use string references in relationships: relationship("ModelName")
Slow queries	Add indexes, use explain(), check eager loading strategy
PostgreSQL schema not found	Set search_path in connection URL or use schema in __table_args__

Debugging Tips

# Enable SQL logging
engine = create_engine("...", echo=True)

# Log only specific queries
import logging
logging.getLogger("sqlalchemy.engine").setLevel(logging.INFO)

# Print compiled query
from sqlalchemy.dialects import postgresql
stmt = select(User).where(User.is_active == True)
print(stmt.compile(dialect=postgresql.dialect()))

# Check object state
from sqlalchemy import inspect
state = inspect(user)
print(f"Transient: {state.transient}")
print(f"Pending: {state.pending}")
print(f"Persistent: {state.persistent}")
print(f"Detached: {state.detached}")

# View all pending changes
print(f"New: {db.new}")
print(f"Dirty: {db.dirty}")
print(f"Deleted: {db.deleted}")

Related Topics

The following topics complement SQLAlchemy development and would make useful additions to your reference collection:

1. Python - Alembic: Database migration management for SQLAlchemy
2. Python - FastAPI: Modern web framework commonly paired with SQLAlchemy
3. Python - Pydantic: Data validation for SQLAlchemy models in APIs
4. Database Patterns: Advanced patterns like CQRS, event sourcing
5. PostgreSQL: Database-specific features and optimisations
6. Python - Async IO: Async patterns for SQLAlchemy 2.0+