Sqlalchemy Code Review

SQLAlchemy Code Review

Quick Reference

Review Checklist

• [ ] Sessions use context managers (with, async with)
• [ ] No session sharing across requests or threads
• [ ] Sessions closed/cleaned up properly
• [ ] relationship() uses appropriate lazy strategy
• [ ] Explicit joinedload/selectinload to avoid N+1
• [ ] No lazy loading in loops (N+1 queries)
• [ ] Using SQLAlchemy 2.0 select() syntax, not legacy query()
• [ ] Bulk operations use bulk_insert/bulk_update, not ORM loops
• [ ] Async sessions use proper async context managers
• [ ] Migrations are reversible with downgrade()
• [ ] Data migrations use op.execute() not ORM models
• [ ] Migration dependencies properly ordered

Gates (SQLAlchemy-specific)

Run once per SQLAlchemy-related finding, after you can anchor `file:line` (see review-verification-protocol) and before the finding ships. If a step’s pass condition is not met, do not assert the finding as written—gather evidence, withdraw, downgrade severity, or rephrase as a question.

Gate 1 — Session scope and lifecycle

Gate 2 — N+1, lazy loading, eager loads

Gate 3 — Migrations (Alembic)

When to Load References

• Reviewing session creation/cleanup → sessions.md
• Reviewing model relationships → relationships.md
• Reviewing database queries → queries.md
• Reviewing Alembic migration files → migrations.md

Review Questions

1. Are all sessions properly managed with context managers? 2. Are relationships configured to avoid N+1 queries? 3. Are queries using SQLAlchemy 2.0 select() syntax? 4. Are all migrations reversible and properly tested?

Migrations

Critical Anti-Patterns

1. Non-Reversible Migrations

Problem: Can't rollback, stuck on failed deploys.

# BAD - no downgrade
"""Add user_role column

Revision ID: abc123
"""

def upgrade():
    op.add_column('users', sa.Column('role', sa.String(50)))

def downgrade():
    pass  # Can't rollback!

# GOOD - reversible migration
def upgrade():
    op.add_column('users', sa.Column('role', sa.String(50), nullable=True))

def downgrade():
    op.drop_column('users', 'role')

2. Not Making New Columns Nullable First

Problem: Migration fails on existing data.

# BAD - adding non-nullable column to existing table
def upgrade():
    # Fails if table has existing rows!
    op.add_column('users', sa.Column('email', sa.String(255), nullable=False))

# GOOD - two-step migration
def upgrade():
    # Step 1: Add nullable column
    op.add_column('users', sa.Column('email', sa.String(255), nullable=True))

# Then in a separate migration after backfilling data:
def upgrade():
    # Step 2: Make it non-nullable
    op.alter_column('users', 'email', nullable=False)

def downgrade():
    op.alter_column('users', 'email', nullable=True)

# BETTER - add with server_default
def upgrade():
    op.add_column(
        'users',
        sa.Column('email', sa.String(255), nullable=False, server_default='')
    )
    # Remove server_default in next migration after cleanup

3. Using ORM Models in Migrations

Problem: Model changes break old migrations.

# BAD - using ORM models directly
from app.models import User  # DON'T!

def upgrade():
    session = Session()
    users = session.query(User).all()  # Model might change!
    for user in users:
        user.email = f"{user.username}@example.com"
    session.commit()

# GOOD - use op.execute with raw SQL
def upgrade():
    op.execute(
        """
        UPDATE users
        SET email = username || '@example.com'
        WHERE email IS NULL
        """
    )

# BETTER - use Core Table for complex operations
from sqlalchemy import table, column, String, Integer

def upgrade():
    users_table = table(
        'users',
        column('id', Integer),
        column('username', String),
        column('email', String)
    )

    connection = op.get_bind()
    users = connection.execute(
        select(users_table.c.id, users_table.c.username)
        .where(users_table.c.email.is_(None))
    ).fetchall()

    for user in users:
        connection.execute(
            update(users_table)
            .where(users_table.c.id == user.id)
            .values(email=f"{user.username}@example.com")
        )

4. Not Handling Concurrent Migrations

Problem: Multiple developers create conflicting migrations.

# BAD - no dependency management
"""Add status column

Revision ID: abc123
Revises: xyz789
"""

# Developer B also based on xyz789 - conflict!
"""Add priority column

Revision ID: def456
Revises: xyz789  # Same parent!
"""

# GOOD - use down_revision properly
# Developer A
"""Add status column

Revision ID: abc123
Revises: xyz789
"""

# Developer B rebases
"""Add priority column

Revision ID: def456
Revises: abc123  # Updated after merge
"""

# BETTER - use alembic branches for long-running features
$ alembic revision -m "feature branch" --branch-label feature_x --depends-on abc123

5. Dangerous DDL Without Transactions

Problem: Partial migrations leave database in broken state.

# BAD - multiple DDL operations without transaction
def upgrade():
    op.create_table('temp_users', ...)
    op.execute("INSERT INTO temp_users SELECT * FROM users")
    op.drop_table('users')  # If this fails, temp_users exists but users is gone!
    op.rename_table('temp_users', 'users')

# GOOD - use batch operations for SQLite
def upgrade():
    with op.batch_alter_table('users') as batch_op:
        batch_op.add_column(sa.Column('new_col', sa.String(50)))
        batch_op.drop_column('old_col')

# PostgreSQL supports transactional DDL
def upgrade():
    # These all happen in a transaction by default
    op.add_column('users', sa.Column('new_col', sa.String(50)))
    op.drop_column('users', 'old_col')

# For operations that can't be in a transaction
def upgrade():
    op.execute("CREATE INDEX CONCURRENTLY idx_users_email ON users(email)")

def downgrade():
    op.execute("DROP INDEX CONCURRENTLY idx_users_email")

6. Not Testing Migrations

Problem: Migrations fail in production.

# BAD - no testing
def upgrade():
    # Hope this works in production!
    op.add_column('users', sa.Column('role', sa.String(50)))

# GOOD - test migrations in CI
# tests/test_migrations.py
import pytest
from alembic import command
from alembic.config import Config

def test_migration_upgrade_downgrade():
    config = Config("alembic.ini")

    # Test upgrade
    command.upgrade(config, "head")

    # Test downgrade
    command.downgrade(config, "-1")

    # Test re-upgrade
    command.upgrade(config, "head")

# BETTER - test with actual data
def test_migration_preserves_data():
    config = Config("alembic.ini")

    # Setup test data
    connection = engine.connect()
    connection.execute(
        "INSERT INTO users (username, email) VALUES ('test', 'test@example.com')"
    )

    # Run migration
    command.upgrade(config, "head")

    # Verify data preserved
    result = connection.execute("SELECT * FROM users WHERE username = 'test'")
    assert result.rowcount == 1

7. Not Using Batch Operations for SQLite

Problem: SQLite doesn't support many ALTER TABLE operations.

# BAD - doesn't work on SQLite
def upgrade():
    op.alter_column('users', 'email', type_=sa.String(512))  # Fails on SQLite!

# GOOD - use batch operations
def upgrade():
    with op.batch_alter_table('users', schema=None) as batch_op:
        batch_op.alter_column('email', type_=sa.String(512))

def downgrade():
    with op.batch_alter_table('users', schema=None) as batch_op:
        batch_op.alter_column('email', type_=sa.String(255))

8. Not Handling Large Data Migrations

Problem: Migration times out or locks table.

# BAD - single UPDATE locks entire table
def upgrade():
    op.execute(
        "UPDATE users SET normalized_email = LOWER(email)"
    )  # Locks millions of rows!

# GOOD - batch updates
def upgrade():
    connection = op.get_bind()
    batch_size = 1000
    offset = 0

    while True:
        result = connection.execute(
            f"""
            UPDATE users
            SET normalized_email = LOWER(email)
            WHERE id IN (
                SELECT id FROM users
                WHERE normalized_email IS NULL
                ORDER BY id
                LIMIT {batch_size} OFFSET {offset}
            )
            """
        )

        if result.rowcount == 0:
            break

        offset += batch_size
        # Sleep to avoid overwhelming the database
        import time
        time.sleep(0.1)

# BETTER - use queue/background job for very large tables
def upgrade():
    # Add column
    op.add_column('users', sa.Column('normalized_email', sa.String(255)))

    # Create background job to populate
    # (Actual backfill happens outside migration)
    pass

9. Not Managing Indexes Properly

Problem: Slow queries after migration, or failed migrations.

# BAD - adding index inline blocks table
def upgrade():
    op.add_column('users', sa.Column('email', sa.String(255)))
    op.create_index('idx_users_email', 'users', ['email'])  # Locks table!

# GOOD - create index concurrently (PostgreSQL)
def upgrade():
    op.add_column('users', sa.Column('email', sa.String(255)))

    # Separate connection for concurrent index
    op.execute("COMMIT")  # End transaction
    op.execute("CREATE INDEX CONCURRENTLY idx_users_email ON users(email)")

def downgrade():
    op.execute("DROP INDEX CONCURRENTLY idx_users_email")
    op.drop_column('users', 'email')

# BETTER - track index creation separately
def upgrade():
    op.add_column('users', sa.Column('email', sa.String(255)))
    # Create index in a separate migration

10. Not Documenting Complex Migrations

Problem: Team doesn't understand migration purpose or impact.

# BAD - no documentation
"""revision abc123
"""

def upgrade():
    op.execute("complex SQL here...")

# GOOD - clear documentation
"""Add normalized_email column for case-insensitive lookups

This migration:
1. Adds a new normalized_email column (nullable initially)
2. Backfills it with lowercase email values
3. Creates a unique index on normalized_email
4. Does NOT make it non-nullable yet (requires follow-up migration)

Expected duration: ~2 minutes for 1M users
Locks: Brief lock during index creation
Rollback safe: Yes

Revision ID: abc123
Revises: xyz789
Create Date: 2024-01-15 10:30:00
"""

def upgrade():
    # Step 1: Add column
    op.add_column(
        'users',
        sa.Column('normalized_email', sa.String(255), nullable=True)
    )

    # Step 2: Backfill in batches
    connection = op.get_bind()
    batch_size = 1000
    # ... batched update logic ...

    # Step 3: Create index
    op.create_index(
        'idx_users_normalized_email',
        'users',
        ['normalized_email'],
        unique=True
    )

def downgrade():
    op.drop_index('idx_users_normalized_email', table_name='users')
    op.drop_column('users', 'normalized_email')

11. Not Using Check Constraints

Problem: Invalid data gets inserted.

# BAD - no constraints, rely on application validation
def upgrade():
    op.add_column('users', sa.Column('age', sa.Integer))

# GOOD - add check constraints
def upgrade():
    op.add_column('users', sa.Column('age', sa.Integer))
    op.create_check_constraint(
        'ck_users_age_positive',
        'users',
        'age >= 0 AND age <= 150'
    )

def downgrade():
    op.drop_constraint('ck_users_age_positive', 'users')
    op.drop_column('users', 'age')

# BETTER - use enum for limited values
from sqlalchemy import Enum

def upgrade():
    role_enum = sa.Enum('user', 'admin', 'moderator', name='user_role')
    role_enum.create(op.get_bind())

    op.add_column(
        'users',
        sa.Column('role', role_enum, nullable=False, server_default='user')
    )

def downgrade():
    op.drop_column('users', 'role')
    sa.Enum(name='user_role').drop(op.get_bind())

Review Questions

1. Does every migration have a working downgrade() function? 2. Are new non-nullable columns added in two steps (nullable first, then constrain)? 3. Are data migrations using op.execute() not ORM models? 4. Are large data updates batched to avoid timeouts? 5. Are indexes created with CONCURRENTLY on PostgreSQL? 6. Are complex migrations documented with expected duration and impact? 7. Are constraints (CHECK, UNIQUE, FK) properly created and dropped?

Queries

Critical Anti-Patterns

1. Using Legacy query() Instead of select()

Problem: Legacy API, deprecated in SQLAlchemy 2.0.

# BAD - legacy query() API (deprecated)
def get_active_users():
    with Session() as session:
        users = session.query(User).filter(User.active == True).all()
        return users

# GOOD - SQLAlchemy 2.0 select() syntax
from sqlalchemy import select

def get_active_users():
    with Session() as session:
        result = session.execute(
            select(User).where(User.active == True)
        )
        return result.scalars().all()

# ASYNC version
async def get_active_users():
    async with AsyncSession() as session:
        result = await session.execute(
            select(User).where(User.active == True)
        )
        return result.scalars().all()

2. Loading Full Objects When Only Columns Needed

Problem: ORM overhead, unnecessary data transfer.

# BAD - loading full ORM objects just for one column
def get_user_emails():
    with Session() as session:
        users = session.execute(select(User)).scalars().all()
        return [user.email for user in users]  # Loaded entire object!

# GOOD - select only needed columns
def get_user_emails():
    with Session() as session:
        result = session.execute(
            select(User.email)
        )
        return result.scalars().all()

# BETTER - multiple columns as tuples
def get_user_info():
    with Session() as session:
        result = session.execute(
            select(User.id, User.name, User.email)
        )
        return result.all()  # Returns list of tuples

3. Using all() When Only One Result Expected

Problem: Confusing API, loads unnecessary data.

# BAD - using all() when expecting one result
def get_user_by_email(email: str):
    with Session() as session:
        users = session.execute(
            select(User).where(User.email == email)
        ).scalars().all()
        return users[0] if users else None  # Awkward!

# GOOD - use scalar_one_or_none()
def get_user_by_email(email: str) -> User | None:
    with Session() as session:
        return session.execute(
            select(User).where(User.email == email)
        ).scalar_one_or_none()

# Use scalar_one() if must exist (raises if not found)
def get_user_by_id(user_id: int) -> User:
    with Session() as session:
        return session.execute(
            select(User).where(User.id == user_id)
        ).scalar_one()  # Raises NoResultFound or MultipleResultsFound

4. Not Using Bulk Operations

Problem: ORM overhead per object, slow inserts/updates.

# BAD - ORM insert in loop
def create_users(user_data: list[dict]):
    with Session() as session:
        for data in user_data:
            user = User(**data)
            session.add(user)  # Individual ORM overhead per user
        session.commit()

# GOOD - bulk insert
def create_users(user_data: list[dict]):
    with Session() as session:
        session.bulk_insert_mappings(User, user_data)
        session.commit()

# BETTER - Core insert for maximum performance
from sqlalchemy import insert

def create_users(user_data: list[dict]):
    with Session() as session:
        session.execute(
            insert(User),
            user_data
        )
        session.commit()

# ASYNC bulk insert
async def create_users(user_data: list[dict]):
    async with AsyncSession() as session:
        await session.execute(
            insert(User),
            user_data
        )
        await session.commit()

5. Not Using Bulk Updates

Problem: ORM overhead, multiple UPDATE statements.

# BAD - update in loop
def deactivate_old_users(cutoff_date):
    with Session() as session:
        users = session.execute(
            select(User).where(User.last_login < cutoff_date)
        ).scalars().all()

        for user in users:
            user.active = False  # Individual UPDATE per user
        session.commit()

# GOOD - single UPDATE statement
from sqlalchemy import update

def deactivate_old_users(cutoff_date):
    with Session() as session:
        session.execute(
            update(User)
            .where(User.last_login < cutoff_date)
            .values(active=False)
        )
        session.commit()

# ASYNC version
async def deactivate_old_users(cutoff_date):
    async with AsyncSession() as session:
        await session.execute(
            update(User)
            .where(User.last_login < cutoff_date)
            .values(active=False)
        )
        await session.commit()

6. Not Using exists() for Existence Checks

Problem: Loads unnecessary data just to check existence.

# BAD - loading data just to check existence
def user_exists(email: str) -> bool:
    with Session() as session:
        user = session.execute(
            select(User).where(User.email == email)
        ).scalar_one_or_none()
        return user is not None  # Loaded entire object!

# GOOD - use exists()
from sqlalchemy import exists, select

def user_exists(email: str) -> bool:
    with Session() as session:
        return session.execute(
            select(exists().where(User.email == email))
        ).scalar()

# Alternative with count (less efficient but sometimes clearer)
from sqlalchemy import func

def user_exists(email: str) -> bool:
    with Session() as session:
        count = session.execute(
            select(func.count()).select_from(User).where(User.email == email)
        ).scalar()
        return count > 0

7. Not Using Pagination

Problem: Memory exhaustion on large result sets.

# BAD - loading all results into memory
def get_all_users():
    with Session() as session:
        users = session.execute(select(User)).scalars().all()  # OOM on millions!
        return users

# GOOD - use limit/offset for pagination
def get_users_page(page: int = 1, page_size: int = 100):
    with Session() as session:
        offset = (page - 1) * page_size
        users = session.execute(
            select(User)
            .offset(offset)
            .limit(page_size)
        ).scalars().all()
        return users

# BETTER - use keyset pagination for large datasets
def get_users_after(last_id: int | None = None, page_size: int = 100):
    with Session() as session:
        query = select(User).order_by(User.id)
        if last_id:
            query = query.where(User.id > last_id)

        users = session.execute(
            query.limit(page_size)
        ).scalars().all()
        return users

# BEST - stream results for very large datasets
def stream_all_users():
    with Session() as session:
        result = session.execute(select(User))
        for user in result.scalars():  # Streams, doesn't load all
            yield user

8. Not Using with_for_update for Row Locking

Problem: Race conditions in concurrent updates.

# BAD - race condition in concurrent requests
def decrement_stock(product_id: int, quantity: int):
    with Session() as session:
        product = session.execute(
            select(Product).where(Product.id == product_id)
        ).scalar_one()

        # Another request could modify stock here!
        if product.stock >= quantity:
            product.stock -= quantity
            session.commit()
        else:
            raise ValueError("Insufficient stock")

# GOOD - use SELECT FOR UPDATE
def decrement_stock(product_id: int, quantity: int):
    with Session() as session:
        with session.begin():
            product = session.execute(
                select(Product)
                .where(Product.id == product_id)
                .with_for_update()  # Row locked until commit
            ).scalar_one()

            if product.stock >= quantity:
                product.stock -= quantity
            else:
                raise ValueError("Insufficient stock")

# ASYNC version
async def decrement_stock(product_id: int, quantity: int):
    async with AsyncSession() as session:
        async with session.begin():
            result = await session.execute(
                select(Product)
                .where(Product.id == product_id)
                .with_for_update()
            )
            product = result.scalar_one()

            if product.stock >= quantity:
                product.stock -= quantity
            else:
                raise ValueError("Insufficient stock")

9. Using String-Based Filters Instead of Column Objects

Problem: No IDE support, error-prone, SQL injection risk.

# BAD - string-based filters
def search_users(name: str):
    with Session() as session:
        users = session.execute(
            select(User).filter_by(name=name)  # String-based
        ).scalars().all()
        return users

# WORSE - string SQL (SQL injection risk!)
def search_users(name: str):
    with Session() as session:
        users = session.execute(
            f"SELECT * FROM users WHERE name = '{name}'"  # NEVER DO THIS!
        ).all()

# GOOD - column object filters
def search_users(name: str):
    with Session() as session:
        users = session.execute(
            select(User).where(User.name == name)  # Type-safe
        ).scalars().all()
        return users

# BETTER - parameterized for complex filters
from sqlalchemy import text

def search_users_complex(filters: dict):
    with Session() as session:
        query = select(User)
        if "name" in filters:
            query = query.where(User.name.contains(filters["name"]))
        if "active" in filters:
            query = query.where(User.active == filters["active"])

        users = session.execute(query).scalars().all()
        return users

10. Not Using Subqueries Efficiently

Problem: Multiple queries instead of single subquery.

# BAD - multiple queries
def get_users_with_recent_posts():
    with Session() as session:
        # First query
        recent_post_user_ids = session.execute(
            select(Post.user_id)
            .where(Post.created_at > datetime.now() - timedelta(days=7))
            .distinct()
        ).scalars().all()

        # Second query
        users = session.execute(
            select(User).where(User.id.in_(recent_post_user_ids))
        ).scalars().all()
        return users

# GOOD - single query with subquery
def get_users_with_recent_posts():
    with Session() as session:
        recent_posts_subq = (
            select(Post.user_id)
            .where(Post.created_at > datetime.now() - timedelta(days=7))
            .distinct()
            .subquery()
        )

        users = session.execute(
            select(User).where(User.id.in_(select(recent_posts_subq.c.user_id)))
        ).scalars().all()
        return users

# BETTER - use join
def get_users_with_recent_posts():
    with Session() as session:
        users = session.execute(
            select(User)
            .join(Post)
            .where(Post.created_at > datetime.now() - timedelta(days=7))
            .distinct()
        ).scalars().all()
        return users

11. Not Using union/union_all

Problem: Multiple queries when one combined query would work.

# BAD - multiple queries
def get_all_content():
    with Session() as session:
        posts = session.execute(select(Post)).scalars().all()
        pages = session.execute(select(Page)).scalars().all()
        return {"posts": posts, "pages": pages}

# GOOD - union query (if columns match)
from sqlalchemy import union_all

def get_all_content_items():
    with Session() as session:
        posts_query = select(
            Post.id,
            Post.title,
            Post.created_at,
            literal("post").label("type")
        )

        pages_query = select(
            Page.id,
            Page.title,
            Page.created_at,
            literal("page").label("type")
        )

        combined = union_all(posts_query, pages_query)
        result = session.execute(combined).all()
        return result

Review Questions

1. Are all queries using SQLAlchemy 2.0 select() syntax not legacy query()? 2. Are bulk operations used for batch inserts/updates? 3. Are only required columns selected when full objects aren't needed? 4. Is exists() used instead of loading objects for existence checks? 5. Is pagination implemented for large result sets? 6. Is with_for_update() used for concurrent updates? 7. Are column objects used instead of string-based filters?

Relationships

Critical Anti-Patterns

1. N+1 Query Problem

Problem: One query per related object, severe performance degradation.

# BAD - N+1 queries
def get_users_with_posts():
    with Session() as session:
        users = session.execute(select(User)).scalars().all()
        result = []
        for user in users:
            # Each access triggers a separate query!
            posts = user.posts  # SELECT * FROM posts WHERE user_id = ?
            result.append({"user": user, "posts": posts})
        return result

# GOOD - eager load with joinedload
from sqlalchemy.orm import joinedload

def get_users_with_posts():
    with Session() as session:
        users = session.execute(
            select(User).options(joinedload(User.posts))
        ).unique().scalars().all()
        return users

# ASYNC version
async def get_users_with_posts():
    async with AsyncSession() as session:
        result = await session.execute(
            select(User).options(joinedload(User.posts))
        )
        return result.unique().scalars().all()

2. Wrong Lazy Loading Strategy

Problem: Default lazy loading causes N+1 in most real-world scenarios.

# BAD - default lazy='select' causes N+1
class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    posts = relationship("Post", back_populates="user")  # lazy='select' by default

# GOOD - choose appropriate lazy strategy
class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)

    # Option 1: lazy='joined' - always join
    posts = relationship("Post", back_populates="user", lazy="joined")

    # Option 2: lazy='selectin' - single extra query
    posts = relationship("Post", back_populates="user", lazy="selectin")

    # Option 3: lazy='raise' - force explicit loading
    posts = relationship("Post", back_populates="user", lazy="raise")

# BEST - use lazy='raise' and explicit loading at query time
class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    posts = relationship("Post", back_populates="user", lazy="raise")

# Then explicitly load when needed
def get_user_with_posts(user_id: int):
    with Session() as session:
        user = session.execute(
            select(User)
            .options(selectinload(User.posts))
            .where(User.id == user_id)
        ).scalar_one()
        return user

3. Missing back_populates

Problem: One-way relationship, inconsistent state, bugs.

# BAD - missing back_populates
class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    posts = relationship("Post")

class Post(Base):
    __tablename__ = "posts"
    id = Column(Integer, primary_key=True)
    user_id = Column(Integer, ForeignKey("users.id"))
    # No relationship back to User!

# GOOD - bidirectional with back_populates
class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    posts = relationship("Post", back_populates="user")

class Post(Base):
    __tablename__ = "posts"
    id = Column(Integer, primary_key=True)
    user_id = Column(Integer, ForeignKey("users.id"))
    user = relationship("User", back_populates="posts")

4. Cascade Not Set Properly

Problem: Orphaned records, foreign key violations.

# BAD - no cascade, orphaned posts when user deleted
class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    posts = relationship("Post", back_populates="user")

# Deleting user leaves orphaned posts or fails with FK constraint

# GOOD - proper cascade for composition
class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    posts = relationship(
        "Post",
        back_populates="user",
        cascade="all, delete-orphan"  # Delete posts when user deleted
    )

# For many-to-many, different cascade
class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    groups = relationship(
        "Group",
        secondary="user_groups",
        back_populates="users",
        cascade="save-update, merge"  # Don't delete groups
    )

5. Using joinedload with Many-to-Many

Problem: Cartesian product explosion, duplicate rows.

# BAD - joinedload with many-to-many causes duplicates
def get_users_with_groups_and_posts():
    with Session() as session:
        users = session.execute(
            select(User)
            .options(joinedload(User.groups))
            .options(joinedload(User.posts))
        ).scalars().all()  # Cartesian product: users × groups × posts!

# GOOD - use selectinload for collections
from sqlalchemy.orm import selectinload

def get_users_with_groups_and_posts():
    with Session() as session:
        users = session.execute(
            select(User)
            .options(selectinload(User.groups))
            .options(selectinload(User.posts))
        ).scalars().all()  # Two separate IN queries, no cartesian product

6. Not Using contains_eager for Filtered Joins

Problem: Inefficient loading when filtering related objects.

# BAD - loads all posts, then filters in Python
def get_users_with_published_posts():
    with Session() as session:
        users = session.execute(
            select(User).options(selectinload(User.posts))
        ).scalars().all()

        # Filters in Python, wasteful
        return [
            {
                "user": user,
                "posts": [p for p in user.posts if p.published]
            }
            for user in users
        ]

# GOOD - use contains_eager with join filter
from sqlalchemy.orm import contains_eager

def get_users_with_published_posts():
    with Session() as session:
        users = session.execute(
            select(User)
            .join(User.posts)
            .where(Post.published == True)
            .options(contains_eager(User.posts))
        ).unique().scalars().all()
        return users

7. Circular Eager Loading

Problem: Infinite recursion with bidirectional relationships.

# BAD - circular eager loading
class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    posts = relationship("Post", back_populates="user", lazy="joined")

class Post(Base):
    __tablename__ = "posts"
    id = Column(Integer, primary_key=True)
    user_id = Column(Integer, ForeignKey("users.id"))
    user = relationship("User", back_populates="posts", lazy="joined")

# Querying User loads Posts which loads User which loads Posts...

# GOOD - one side lazy, or explicit loading
class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    posts = relationship("Post", back_populates="user", lazy="raise")

class Post(Base):
    __tablename__ = "posts"
    id = Column(Integer, primary_key=True)
    user_id = Column(Integer, ForeignKey("users.id"))
    user = relationship("User", back_populates="posts", lazy="raise")

# Explicitly load what you need
def get_user_with_posts(user_id: int):
    with Session() as session:
        return session.execute(
            select(User)
            .options(selectinload(User.posts))
            .where(User.id == user_id)
        ).scalar_one()

8. Not Using Association Object for Rich M2M

Problem: Can't store additional attributes on join table.

# BAD - simple secondary table, can't add attributes
user_groups = Table(
    "user_groups",
    Base.metadata,
    Column("user_id", ForeignKey("users.id")),
    Column("group_id", ForeignKey("groups.id"))
)

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    groups = relationship("Group", secondary=user_groups)

# Can't store "joined_at" or "role" on the relationship!

# GOOD - association object pattern
class UserGroup(Base):
    __tablename__ = "user_groups"
    user_id = Column(Integer, ForeignKey("users.id"), primary_key=True)
    group_id = Column(Integer, ForeignKey("groups.id"), primary_key=True)
    joined_at = Column(DateTime, default=datetime.utcnow)
    role = Column(String)  # "admin", "member", etc.

    user = relationship("User", back_populates="group_associations")
    group = relationship("Group", back_populates="user_associations")

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    group_associations = relationship("UserGroup", back_populates="user")

    # Convenience property
    @property
    def groups(self):
        return [assoc.group for assoc in self.group_associations]

class Group(Base):
    __tablename__ = "groups"
    id = Column(Integer, primary_key=True)
    user_associations = relationship("UserGroup", back_populates="group")

9. Not Using raiseload for Debugging

Problem: N+1 queries slip into production unnoticed.

# BAD - lazy loading hidden issues in production
from sqlalchemy.orm import Session

def get_users():
    with Session() as session:
        users = session.execute(select(User)).scalars().all()
        # Accessing posts triggers lazy load - silent N+1 in production
        for user in users:
            print(user.posts)

# GOOD - use raiseload in development to catch issues
from sqlalchemy.orm import raiseload

def get_users():
    with Session() as session:
        users = session.execute(
            select(User).options(raiseload("*"))  # Raise on any lazy load
        ).scalars().all()
        # This will raise immediately, forcing us to fix it
        for user in users:
            print(user.posts)  # InvalidRequestError!

# FIX - explicit loading
def get_users():
    with Session() as session:
        users = session.execute(
            select(User).options(selectinload(User.posts))
        ).scalars().all()
        for user in users:
            print(user.posts)  # No lazy load, efficient!

Review Questions

1. Are all relationship queries using explicit eager loading (joinedload, selectinload)? 2. Is lazy='raise' used to prevent accidental lazy loading? 3. Do all relationships have proper back_populates? 4. Are cascade options set appropriately for composition vs association? 5. Is selectinload used instead of joinedload for collections? 6. Are association objects used for many-to-many with attributes?

Sessions

Critical Anti-Patterns

1. Session Not Closed

Problem: Connection pool exhaustion, memory leaks.

# BAD - session never closed
def get_user(user_id: int):
    session = Session()
    user = session.get(User, user_id)
    return user  # Session leaked!

# GOOD - using context manager
def get_user(user_id: int) -> User | None:
    with Session() as session:
        user = session.get(User, user_id)
        return user

2. Session Shared Across Requests

Problem: Concurrent modifications, race conditions, data corruption.

# BAD - global session shared across requests
session = Session()  # Module-level!

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    user = session.get(User, user_id)  # Multiple requests share session!
    return user

# GOOD - request-scoped session
from contextlib import asynccontextmanager

@asynccontextmanager
async def get_db_session():
    async with AsyncSession() as session:
        try:
            yield session
        finally:
            await session.close()

@app.get("/users/{user_id}")
async def get_user(user_id: int, session = Depends(get_db_session)):
    user = await session.get(User, user_id)
    return user

3. Manual Commit Without Rollback Handling

Problem: Partial commits, inconsistent state on errors.

# BAD - no rollback on error
def create_user(name: str, email: str):
    session = Session()
    user = User(name=name, email=email)
    session.add(user)
    session.commit()  # If this fails, session corrupted
    session.close()
    return user

# GOOD - proper error handling
def create_user(name: str, email: str) -> User:
    with Session() as session:
        try:
            user = User(name=name, email=email)
            session.add(user)
            session.commit()
            return user
        except Exception:
            session.rollback()
            raise

4. Using Sync Session in Async Context

Problem: Blocks event loop, poor performance.

# BAD - blocking sync session in async
from sqlalchemy.orm import Session

async def get_user(user_id: int):
    with Session() as session:  # Blocks event loop!
        user = session.get(User, user_id)
        return user

# GOOD - async session
from sqlalchemy.ext.asyncio import AsyncSession

async def get_user(user_id: int) -> User | None:
    async with AsyncSession() as session:
        result = await session.execute(
            select(User).where(User.id == user_id)
        )
        return result.scalar_one_or_none()

5. Session Used After Commit

Problem: DetachedInstanceError, expired objects.

# BAD - accessing object after session closed
def get_user_data(user_id: int):
    with Session() as session:
        user = session.get(User, user_id)
    return user.email  # DetachedInstanceError! Session closed

# GOOD - access data before session closes
def get_user_data(user_id: int) -> str | None:
    with Session() as session:
        user = session.get(User, user_id)
        if user:
            return user.email
        return None

# BETTER - use expunge or eager loading
from sqlalchemy.orm import joinedload

def get_user_with_posts(user_id: int) -> User | None:
    with Session() as session:
        user = session.execute(
            select(User)
            .options(joinedload(User.posts))
            .where(User.id == user_id)
        ).scalar_one_or_none()

        if user:
            session.expunge(user)  # Detach from session
        return user

6. Not Using Session.begin() for Transactions

Problem: AutoCommit confusion, no explicit transaction boundaries.

# BAD - implicit transaction boundaries
def transfer_money(from_id: int, to_id: int, amount: float):
    with Session() as session:
        from_account = session.get(Account, from_id)
        to_account = session.get(Account, to_id)

        from_account.balance -= amount
        session.commit()  # First commit

        to_account.balance += amount
        session.commit()  # Second commit - money lost if this fails!

# GOOD - explicit transaction with begin()
def transfer_money(from_id: int, to_id: int, amount: float):
    with Session() as session:
        with session.begin():
            from_account = session.get(Account, from_id)
            to_account = session.get(Account, to_id)

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

            from_account.balance -= amount
            to_account.balance += amount
            # Both committed together or rolled back together

# ASYNC version
async def transfer_money(from_id: int, to_id: int, amount: float):
    async with AsyncSession() as session:
        async with session.begin():
            result = await session.execute(
                select(Account).where(Account.id.in_([from_id, to_id]))
            )
            accounts = {acc.id: acc for acc in result.scalars()}

            from_account = accounts[from_id]
            to_account = accounts[to_id]

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

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

7. Session Factory Not Configured Properly

Problem: Inconsistent session behavior, connection issues.

# BAD - new engine every time
def get_session():
    engine = create_engine("postgresql://...")  # New engine each call!
    Session = sessionmaker(bind=engine)
    return Session()

# GOOD - reuse engine and session factory
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# Module level - create once
engine = create_engine(
    "postgresql://...",
    pool_pre_ping=True,  # Verify connections
    pool_size=10,
    max_overflow=20
)

SessionLocal = sessionmaker(
    bind=engine,
    expire_on_commit=False,  # Don't expire objects on commit
    autocommit=False,
    autoflush=False
)

def get_session():
    return SessionLocal()

# ASYNC version
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession

async_engine = create_async_engine(
    "postgresql+asyncpg://...",
    pool_pre_ping=True,
    pool_size=10
)

AsyncSessionLocal = sessionmaker(
    async_engine,
    class_=AsyncSession,
    expire_on_commit=False,
    autocommit=False,
    autoflush=False
)

async def get_async_session():
    async with AsyncSessionLocal() as session:
        yield session

8. Missing Session Refresh After Background Operations

Problem: Stale data when session persists across long operations.

# BAD - using stale session data
async def process_order(order_id: int):
    async with AsyncSession() as session:
        order = await session.get(Order, order_id)

        # Long running background task
        await process_payment(order.id)  # Another process might update order

        # order.status might be stale here!
        if order.status == "pending":
            order.status = "completed"
            await session.commit()

# GOOD - refresh after external operations
async def process_order(order_id: int):
    async with AsyncSession() as session:
        order = await session.get(Order, order_id)

        await process_payment(order.id)

        # Refresh to get latest state
        await session.refresh(order)

        if order.status == "pending":
            order.status = "completed"
            await session.commit()

Review Questions

1. Are all sessions using context managers (with or async with)? 2. Is each request/thread getting its own session instance? 3. Are transactions using explicit session.begin()? 4. Are async contexts using AsyncSession not sync Session? 5. Are objects accessed before the session closes? 6. Is the session factory configured once and reused?