Sqlalchemy

SQLAlchemy ORM Skill

--- progressive_disclosure: entry_point: summary: "Python SQL toolkit and ORM with powerful query builder and relationship mapping" when_to_use:

• "When building Python applications with databases"
• "When needing complex SQL queries with type safety"
• "When working with FastAPI/Flask/Django"
• "When needing database migrations (Alembic)"

quick_start:

• "pip install sqlalchemy"
• "Define models with declarative base"
• "Create engine and session"
• "Query with select() and commit()"

token_estimate: entry: 70-85 full: 4500-5500 ---

Core Concepts

SQLAlchemy 2.0 Modern API

SQLAlchemy 2.0 introduced modern patterns with better type hints, improved query syntax, and async support.

Key Changes from 1.x:

• select() instead of Query
• Mapped[T] and mapped_column() for type hints
• Explicit Session.execute() for queries
• Better async support with AsyncSession

Installation

# Core SQLAlchemy
pip install sqlalchemy

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

# With Alembic for migrations
pip install alembic

# FastAPI integration
pip install fastapi sqlalchemy

Declarative Models (SQLAlchemy 2.0)

Basic Model Definition

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

# Base class for all models
class Base(DeclarativeBase):
    pass

# User model with type hints
class User(Base):
    __tablename__ = "users"

    # Primary key
    id: Mapped[int] = mapped_column(primary_key=True)

    # Required fields
    email: Mapped[str] = mapped_column(String(255), unique=True, index=True)
    username: Mapped[str] = mapped_column(String(50), unique=True)
    hashed_password: Mapped[str] = mapped_column(String(255))

    # Optional fields
    full_name: Mapped[Optional[str]] = mapped_column(String(100))
    is_active: Mapped[bool] = mapped_column(default=True)

    # Timestamps with server defaults
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        server_default=func.now()
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        server_default=func.now(),
        onupdate=func.now()
    )

    # Relationships
    posts: Mapped[list["Post"]] = relationship(back_populates="author")

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

Relationships

One-to-Many:

class Post(Base):
    __tablename__ = "posts"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    content: Mapped[str]
    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))

    # Relationship with back_populates
    author: Mapped["User"] = relationship(back_populates="posts")
    tags: Mapped[list["Tag"]] = relationship(
        secondary="post_tags",
        back_populates="posts"
    )

Many-to-Many:

from sqlalchemy import Table, Column, Integer, ForeignKey

# Association table
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 Tag(Base):
    __tablename__ = "tags"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50), unique=True)

    posts: Mapped[list["Post"]] = relationship(
        secondary=post_tags,
        back_populates="tags"
    )

Database Setup

Engine and Session Configuration

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session
from sqlalchemy.pool import QueuePool

# Database URL formats
# SQLite: sqlite:///./database.db
# PostgreSQL: postgresql://user:pass@localhost/dbname  # pragma: allowlist secret
# MySQL: mysql+pymysql://user:pass@localhost/dbname  # pragma: allowlist secret

DATABASE_URL = "postgresql://user:pass@localhost/mydb"  # pragma: allowlist secret

# Create engine with connection pooling
engine = create_engine(
    DATABASE_URL,
    poolclass=QueuePool,
    pool_size=5,
    max_overflow=10,
    pool_pre_ping=True,  # Check connection before using
    echo=False  # Set True for SQL logging
)

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

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

Dependency Injection (FastAPI Pattern)

from typing import Generator

def get_db() -> Generator[Session, None, None]:
    """Database session dependency for FastAPI."""
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# Usage in FastAPI
from fastapi import Depends

@app.get("/users/{user_id}")
def get_user(user_id: int, db: Session = Depends(get_db)):
    return db.execute(
        select(User).where(User.id == user_id)
    ).scalar_one_or_none()

Query Patterns (SQLAlchemy 2.0)

Select Queries

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

# Basic select
stmt = select(User).where(User.email == "user@example.com")
user = session.execute(stmt).scalar_one_or_none()

# Multiple conditions
stmt = select(User).where(
    and_(
        User.is_active == True,
        User.created_at > datetime(2024, 1, 1)
    )
)
users = session.execute(stmt).scalars().all()

# OR conditions
stmt = select(User).where(
    or_(
        User.email.like("%@gmail.com"),
        User.email.like("%@yahoo.com")
    )
)

# Ordering and limiting
stmt = (
    select(User)
    .where(User.is_active == True)
    .order_by(desc(User.created_at))
    .limit(10)
    .offset(20)
)

# Counting
stmt = select(func.count()).select_from(User)
count = session.execute(stmt).scalar()

Joins

# Inner join
stmt = (
    select(Post, User)
    .join(User, Post.user_id == User.id)
    .where(User.is_active == True)
)
results = session.execute(stmt).all()

# Left outer join
stmt = (
    select(User, func.count(Post.id).label("post_count"))
    .outerjoin(Post)
    .group_by(User.id)
)

# Multiple joins
stmt = (
    select(Post)
    .join(Post.author)
    .join(Post.tags)
    .where(Tag.name == "python")
)

Eager Loading (Solve N+1 Problem)

from sqlalchemy.orm import selectinload, joinedload

# selectinload - separate query (better for collections)
stmt = select(User).options(selectinload(User.posts))
users = session.execute(stmt).scalars().all()
# Now users[0].posts won't trigger additional queries

# joinedload - single query with join (better for one-to-one)
stmt = select(Post).options(joinedload(Post.author))
posts = session.execute(stmt).unique().scalars().all()

# Nested eager loading
stmt = select(User).options(
    selectinload(User.posts).selectinload(Post.tags)
)

# Load only specific columns
from sqlalchemy.orm import load_only
stmt = select(User).options(load_only(User.id, User.email))

CRUD Operations

Create

def create_user(db: Session, email: str, username: str, password: str):
    """Create new user."""
    user = User(
        email=email,
        username=username,
        hashed_password=hash_password(password)
    )
    db.add(user)
    db.commit()
    db.refresh(user)  # Get updated fields (id, timestamps)
    return user

# Bulk insert
users = [
    User(email=f"user{i}@example.com", username=f"user{i}")
    for i in range(100)
]
db.add_all(users)
db.commit()

Read

def get_user_by_email(db: Session, email: str) -> Optional[User]:
    """Get user by email."""
    stmt = select(User).where(User.email == email)
    return db.execute(stmt).scalar_one_or_none()

def get_users(
    db: Session,
    skip: int = 0,
    limit: int = 100
) -> list[User]:
    """Get paginated users."""
    stmt = (
        select(User)
        .where(User.is_active == True)
        .order_by(User.created_at.desc())
        .offset(skip)
        .limit(limit)
    )
    return db.execute(stmt).scalars().all()

Update

def update_user(db: Session, user_id: int, **kwargs):
    """Update user fields."""
    stmt = select(User).where(User.id == user_id)
    user = db.execute(stmt).scalar_one_or_none()

    if not user:
        return None

    for key, value in kwargs.items():
        setattr(user, key, value)

    db.commit()
    db.refresh(user)
    return user

# Bulk update
from sqlalchemy import update

stmt = (
    update(User)
    .where(User.is_active == False)
    .values(deleted_at=datetime.utcnow())
)
db.execute(stmt)
db.commit()

Delete

def delete_user(db: Session, user_id: int) -> bool:
    """Delete user."""
    stmt = select(User).where(User.id == user_id)
    user = db.execute(stmt).scalar_one_or_none()

    if not user:
        return False

    db.delete(user)
    db.commit()
    return True

# Bulk delete
from sqlalchemy import delete

stmt = delete(User).where(User.is_active == False)
db.execute(stmt)
db.commit()

Transactions and Session Management

Context Manager Pattern

from contextlib import contextmanager

@contextmanager
def get_db_session():
    """Session context manager."""
    session = SessionLocal()
    try:
        yield session
        session.commit()
    except Exception:
        session.rollback()
        raise
    finally:
        session.close()

# Usage
with get_db_session() as db:
    user = create_user(db, "test@example.com", "testuser", "password")
    # Auto-commits on success, rollback on exception

Manual Transaction Control

def transfer_money(db: Session, from_user_id: int, to_user_id: int, amount: float):
    """Transfer money between users with transaction."""
    try:
        # Begin nested transaction
        with db.begin_nested():
            # Deduct from sender
            stmt = select(User).where(User.id == from_user_id).with_for_update()
            sender = db.execute(stmt).scalar_one()
            sender.balance -= amount

            # Add to receiver
            stmt = select(User).where(User.id == to_user_id).with_for_update()
            receiver = db.execute(stmt).scalar_one()
            receiver.balance += amount

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

Async SQLAlchemy

Async Setup

from sqlalchemy.ext.asyncio import (
    create_async_engine,
    AsyncSession,
    async_sessionmaker
)

# Async engine (note: asyncpg for PostgreSQL, aiosqlite for SQLite)
DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/mydb"  # pragma: allowlist secret

async_engine = create_async_engine(
    DATABASE_URL,
    echo=False,
    pool_size=5,
    max_overflow=10
)

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

# Create tables
async def init_db():
    async with async_engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

Async CRUD Operations

async def get_user_async(user_id: int) -> Optional[User]:
    """Get user asynchronously."""
    async with AsyncSessionLocal() as session:
        stmt = select(User).where(User.id == user_id)
        result = await session.execute(stmt)
        return result.scalar_one_or_none()

async def create_user_async(email: str, username: str) -> User:
    """Create user asynchronously."""
    async with AsyncSessionLocal() as session:
        user = User(email=email, username=username)
        session.add(user)
        await session.commit()
        await session.refresh(user)
        return user

# FastAPI async dependency
async def get_async_db():
    async with AsyncSessionLocal() as session:
        yield session

@app.get("/users/{user_id}")
async def get_user_endpoint(
    user_id: int,
    db: AsyncSession = Depends(get_async_db)
):
    stmt = select(User).where(User.id == user_id)
    result = await db.execute(stmt)
    return result.scalar_one_or_none()

Alembic Migrations

Setup Alembic

# Initialize Alembic
alembic init alembic

# Edit alembic.ini - set database URL
# sqlalchemy.url = postgresql://user:pass@localhost/mydb  # pragma: allowlist secret

# Or use env variable in alembic/env.py

Configure Alembic

# alembic/env.py
from sqlalchemy import engine_from_config, pool
from alembic import context
from myapp.models import Base  # Import your Base

# Add your model's MetaData
target_metadata = Base.metadata

def run_migrations_online():
    """Run migrations in 'online' mode."""
    configuration = config.get_section(config.config_ini_section)
    configuration["sqlalchemy.url"] = os.getenv("DATABASE_URL")

    connectable = engine_from_config(
        configuration,
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )

    with connectable.connect() as connection:
        context.configure(
            connection=connection,
            target_metadata=target_metadata
        )

        with context.begin_transaction():
            context.run_migrations()

Create and Apply Migrations

# Auto-generate migration from model changes
alembic revision --autogenerate -m "Add users table"

# Review generated migration in alembic/versions/

# Apply migration
alembic upgrade head

# Rollback one version
alembic downgrade -1

# Show current version
alembic current

# Show migration history
alembic history

Manual Migration Example

# alembic/versions/xxx_add_users.py
from alembic import op
import sqlalchemy as sa

def upgrade():
    op.create_table(
        'users',
        sa.Column('id', sa.Integer(), nullable=False),
        sa.Column('email', sa.String(255), nullable=False),
        sa.Column('username', sa.String(50), nullable=False),
        sa.PrimaryKeyConstraint('id')
    )
    op.create_index('ix_users_email', 'users', ['email'], unique=True)

def downgrade():
    op.drop_index('ix_users_email', table_name='users')
    op.drop_table('users')

FastAPI Integration

Complete FastAPI Example

from fastapi import FastAPI, Depends, HTTPException, status
from sqlalchemy.orm import Session
from pydantic import BaseModel, EmailStr
from typing import List

app = FastAPI()

# Pydantic schemas
class UserBase(BaseModel):
    email: EmailStr
    username: str

class UserCreate(UserBase):
    password: str

class UserResponse(UserBase):
    id: int
    is_active: bool
    created_at: datetime

    class Config:
        from_attributes = True  # SQLAlchemy 2.0 (was orm_mode)

# CRUD operations
@app.post("/users/", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
def create_user_endpoint(user: UserCreate, db: Session = Depends(get_db)):
    # Check if user exists
    stmt = select(User).where(User.email == user.email)
    if db.execute(stmt).scalar_one_or_none():
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Email already registered"
        )

    # Create user
    db_user = User(
        email=user.email,
        username=user.username,
        hashed_password=hash_password(user.password)
    )
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

@app.get("/users/{user_id}", response_model=UserResponse)
def read_user(user_id: int, db: Session = Depends(get_db)):
    stmt = select(User).where(User.id == user_id)
    user = db.execute(stmt).scalar_one_or_none()

    if not user:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="User not found"
        )
    return user

@app.get("/users/", response_model=List[UserResponse])
def list_users(
    skip: int = 0,
    limit: int = 100,
    db: Session = Depends(get_db)
):
    stmt = (
        select(User)
        .where(User.is_active == True)
        .offset(skip)
        .limit(limit)
    )
    return db.execute(stmt).scalars().all()

@app.put("/users/{user_id}", response_model=UserResponse)
def update_user(
    user_id: int,
    user_update: UserBase,
    db: Session = Depends(get_db)
):
    stmt = select(User).where(User.id == user_id)
    db_user = db.execute(stmt).scalar_one_or_none()

    if not db_user:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="User not found"
        )

    db_user.email = user_update.email
    db_user.username = user_update.username
    db.commit()
    db.refresh(db_user)
    return db_user

@app.delete("/users/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_user(user_id: int, db: Session = Depends(get_db)):
    stmt = select(User).where(User.id == user_id)
    db_user = db.execute(stmt).scalar_one_or_none()

    if not db_user:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="User not found"
        )

    db.delete(db_user)
    db.commit()

Testing with Pytest

Test Database Setup

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

# In-memory SQLite for testing
SQLALCHEMY_TEST_DATABASE_URL = "sqlite:///:memory:"

@pytest.fixture(scope="function")
def db_session():
    """Create test database session."""
    engine = create_engine(
        SQLALCHEMY_TEST_DATABASE_URL,
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,
    )

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

    TestingSessionLocal = sessionmaker(
        autocommit=False,
        autoflush=False,
        bind=engine
    )

    session = TestingSessionLocal()
    try:
        yield session
    finally:
        session.close()
        Base.metadata.drop_all(bind=engine)

@pytest.fixture(scope="function")
def test_user(db_session):
    """Create test user."""
    user = User(
        email="test@example.com",
        username="testuser",
        hashed_password="hashed"  # pragma: allowlist secret
    )
    db_session.add(user)
    db_session.commit()
    db_session.refresh(user)
    return user

Test Examples

def test_create_user(db_session):
    """Test user creation."""
    user = User(email="new@example.com", username="newuser")
    db_session.add(user)
    db_session.commit()

    assert user.id is not None
    assert user.email == "new@example.com"
    assert user.created_at is not None

def test_query_user(db_session, test_user):
    """Test user query."""
    stmt = select(User).where(User.email == "test@example.com")
    found_user = db_session.execute(stmt).scalar_one()

    assert found_user.id == test_user.id
    assert found_user.username == test_user.username

def test_update_user(db_session, test_user):
    """Test user update."""
    test_user.username = "updated"
    db_session.commit()

    stmt = select(User).where(User.id == test_user.id)
    updated_user = db_session.execute(stmt).scalar_one()
    assert updated_user.username == "updated"

def test_delete_user(db_session, test_user):
    """Test user deletion."""
    user_id = test_user.id
    db_session.delete(test_user)
    db_session.commit()

    stmt = select(User).where(User.id == user_id)
    assert db_session.execute(stmt).scalar_one_or_none() is None

Performance Optimization

Query Optimization

# Use indexes
class User(Base):
    __tablename__ = "users"

    email: Mapped[str] = mapped_column(String(255), index=True, unique=True)
    created_at: Mapped[datetime] = mapped_column(index=True)

    # Composite index
    __table_args__ = (
        Index('ix_user_email_active', 'email', 'is_active'),
    )

# Use select_from for complex queries
stmt = (
    select(func.count(Post.id))
    .select_from(User)
    .join(Post)
    .where(User.is_active == True)
)

# Use contains_eager for joined loads
from sqlalchemy.orm import contains_eager

stmt = (
    select(Post)
    .join(Post.author)
    .options(contains_eager(Post.author))
    .where(User.is_active == True)
)

Connection Pooling

# Configure pool
engine = create_engine(
    DATABASE_URL,
    pool_size=20,           # Number of connections to keep
    max_overflow=10,        # Additional connections when pool full
    pool_timeout=30,        # Seconds to wait for connection
    pool_recycle=3600,      # Recycle connections after 1 hour
    pool_pre_ping=True      # Verify connections before use
)

# Monitor pool
from sqlalchemy import event

@event.listens_for(engine, "connect")
def receive_connect(dbapi_conn, connection_record):
    print("New connection established")

@event.listens_for(engine, "checkout")
def receive_checkout(dbapi_conn, connection_record, connection_proxy):
    print("Connection checked out from pool")

Batch Operations

# Bulk insert with executemany
from sqlalchemy import insert

data = [
    {"email": f"user{i}@example.com", "username": f"user{i}"}
    for i in range(1000)
]

stmt = insert(User)
db.execute(stmt, data)
db.commit()

# Bulk update
from sqlalchemy import update

stmt = (
    update(User)
    .where(User.is_active == False)
    .values(deleted_at=func.now())
)
db.execute(stmt)

Best Practices

1. Use SQLAlchemy 2.0 Syntax: Modern API with better type hints 2. Type Annotations: Use Mapped[T] and mapped_column() 3. Eager Loading: Solve N+1 queries with selectinload/joinedload 4. Session Management: Use dependency injection pattern 5. Migrations: Always use Alembic for schema changes 6. Indexes: Add indexes for frequently queried columns 7. Connection Pooling: Configure appropriate pool settings 8. Testing: Use in-memory SQLite for fast tests 9. Async: Use AsyncSession for async frameworks 10. Error Handling: Always handle NoResultFound and MultipleResultsFound

Common Patterns

Repository Pattern

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

T = TypeVar('T', bound=Base)

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

    def get(self, id: int) -> Optional[T]:
        stmt = select(self.model).where(self.model.id == id)
        return self.db.execute(stmt).scalar_one_or_none()

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

    def create(self, obj: T) -> T:
        self.db.add(obj)
        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

# Usage
user_repo = BaseRepository(User, db)
user = user_repo.get(1)

Soft Delete Pattern

class SoftDeleteMixin:
    deleted_at: Mapped[Optional[datetime]] = mapped_column(default=None)

    @property
    def is_deleted(self) -> bool:
        return self.deleted_at is not None

class User(Base, SoftDeleteMixin):
    __tablename__ = "users"
    # ... fields

# Query only active records
stmt = select(User).where(User.deleted_at.is_(None))

# Soft delete
user.deleted_at = datetime.utcnow()
db.commit()

Audit Trail Pattern

class AuditMixin:
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        server_default=func.now()
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        server_default=func.now(),
        onupdate=func.now()
    )
    created_by: Mapped[Optional[int]] = mapped_column(ForeignKey("users.id"))
    updated_by: Mapped[Optional[int]] = mapped_column(ForeignKey("users.id"))

class Post(Base, AuditMixin):
    __tablename__ = "posts"
    # ... fields

SQL Quality & Efficiency Anti-Patterns

The ORM makes several classic SQL defects easy to introduce by accident. Watch for these during query review:

• Query-in-loop (N+1) — eager-load with selectinload/joinedload instead of one

query per row.

• *`SELECT ` / over-fetching** — project only the columns you use with

select(Model.col_a, Model.col_b).

• Missing indexes / `SELECT DISTINCT` to mask duplicates — index filter/join/order

columns; DISTINCT usually hides a missing join, not a fix.

• Cursor-in-loop writes — replace row-by-row writes with a single set-based

update()/insert().

• DDL/DML interleaving — keep schema changes in migrations; runtime code is DML-only.
• Unparameterized queries — bind parameters; never f-string untrusted data into SQL.
• Correlated subqueries — prefer independent subqueries/CTEs or joins with indexes.

See [sql-quality-antipatterns.md](references/sql-quality-antipatterns.md) for non-compliant vs compliant examples (raw SQL and SQLAlchemy 2.0) and how to test each.

Derived from CAST Highlight SQL code-quality indicators

(https://doc.casthighlight.com/), cross-referenced with the SQL standard and SonarSource

RSPEC. For DDL hygiene see the database-migration skill.

Resources

• SQLAlchemy 2.0 Documentation
• Alembic Documentation
• FastAPI SQLAlchemy Guide
• SQLAlchemy Type Annotations

Related Skills

When using Sqlalchemy, these skills enhance your workflow:

• django: Django ORM patterns and migration strategies for comparison
• test-driven-development: TDD patterns for database models and queries
• fastapi-local-dev: FastAPI + SQLAlchemy integration patterns
• systematic-debugging: Advanced debugging for ORM query issues and N+1 problems

[Full documentation available in these skills if deployed in your bundle]

SQL Quality & Efficiency Anti-Patterns (SQLAlchemy)

High-signal SQL anti-patterns to watch for when writing queries — directly or through the SQLAlchemy ORM. The ORM makes several of these easier to introduce by accident (N+1 query-in-loop, over-fetching, SELECT *-equivalents), so they are worth an explicit review pass.

Source note: These patterns are derived from CAST Highlight's SQL code-quality

indicators (https://doc.casthighlight.com/), cross-referenced with the SQL standard and

SonarSource RSPEC. CAST defers to those primary sources for the underlying rules; where

CAST's own SQL pages had title/content mismatches, the well-established form of the rule

is used. Examples are original and shown in both raw SQL and SQLAlchemy 2.0 form.

This is a deliberately small, high-value subset — not all of CAST's ~49 SQL rules.

---

1. Avoid query-in-loop (the N+1 problem)

Why: Issuing one query per row of a previous result multiplies round-trips. With N parent rows you pay 1 + N queries; latency dominates and the database cache thrashes. This is the single most common ORM performance defect.

Non-compliant (N+1):

authors = session.scalars(select(Author)).all()
for author in authors:
    # one SELECT per author — N+1 round-trips
    books = session.scalars(
        select(Book).where(Book.author_id == author.id)
    ).all()
    print(author.name, len(books))

Compliant — eager-load in one round-trip:

from sqlalchemy.orm import selectinload

authors = session.scalars(
    select(Author).options(selectinload(Author.books))
).all()
for author in authors:
    print(author.name, len(author.books))   # no extra queries

How to test: Wrap the call in a query counter (e.g. SQLAlchemy's event.listen(engine, "before_cursor_execute", ...) or pytest's sqlalchemy.testing.assert_compile/a counter fixture) and assert the number of emitted statements is constant regardless of row count:

def test_authors_with_books_is_constant_queries(query_counter):
    load_authors_with_books(session)
    assert query_counter.count <= 2   # one for authors, one for the selectin load

---

2. Don't SELECT * / over-fetch columns

Why: Selecting every column (or hydrating full ORM entities when you need two fields) pulls unused data over the wire, defeats covering indexes, and breaks silently when the schema changes. Fetch only the columns you use.

Non-compliant:

SELECT * FROM orders WHERE status = 'open';

orders = session.scalars(select(Order).where(Order.status == "open")).all()
total = sum(o.amount for o in orders)   # only needed `amount`, hydrated whole rows

Compliant — project the columns you need:

SELECT id, amount FROM orders WHERE status = 'open';

rows = session.execute(
    select(Order.id, Order.amount).where(Order.status == "open")
).all()
total = sum(amount for _id, amount in rows)

How to test: Assert the compiled SQL names explicit columns rather than */all entity columns:

def test_open_orders_projects_only_needed_columns():
    stmt = open_orders_amount_query()
    sql = str(stmt.compile())
    assert "amount" in sql and "SELECT orders." not in sql.replace("orders.id", "")

---

3. Add indexes for filter/join/order columns; avoid SELECT DISTINCT to mask dupes

Why: A WHERE, JOIN, or ORDER BY on an unindexed column forces a full scan. Reaching for SELECT DISTINCT to remove duplicate rows is usually a symptom of a missing join condition or a missing index, not a fix — it adds a sort/hash on top of the scan.

Non-compliant:

SELECT DISTINCT customer_id FROM orders WHERE region = 'EU';   -- region not indexed

Compliant — index the predicate, drop the DISTINCT if the query is correct:

class Order(Base):
    __tablename__ = "orders"
    id: Mapped[int] = mapped_column(primary_key=True)
    customer_id: Mapped[int] = mapped_column(index=True)
    region: Mapped[str] = mapped_column(index=True)   # predicate column indexed

SELECT customer_id FROM orders WHERE region = 'EU' GROUP BY customer_id;

How to test: In a test database with representative data, capture the query plan (EXPLAIN) and assert it uses an index scan rather than a sequential/full scan for the hot query. For correctness, assert the de-duplicated result matches the intended set.

---

4. Avoid cursor-in-loop; operate set-at-a-time

Why: Row-by-row cursor iteration that issues a write (or another query) per row ("RBAR" — row by agonizing row) serializes work the database could do in one set-based statement. Prefer a single bulk UPDATE/INSERT ... SELECT.

Non-compliant:

for user in session.scalars(select(User).where(User.inactive)):
    user.archived = True
    session.flush()        # a write per row

Compliant — one set-based statement:

from sqlalchemy import update

session.execute(
    update(User).where(User.inactive.is_(True)).values(archived=True)
)

How to test: Count emitted UPDATE statements (query-counter fixture) and assert it is 1 regardless of how many rows match; assert the post-condition (all matching rows archived) with a follow-up SELECT COUNT.

---

5. Don't interleave DDL and DML; keep schema changes separate

Why: Mixing schema changes (CREATE/ALTER/DROP) with data manipulation (INSERT/UPDATE/DELETE) inside one procedure or transaction can force statement recompilation, invalidate cached plans, and on some engines silently auto-commit — making the unit non-atomic. Put DDL in migrations; keep runtime code DML-only.

Non-compliant: a runtime routine that ALTER TABLE ... ADD COLUMN then immediately UPDATEs using the new column.

Compliant: the column is added by an Alembic migration; application code only reads and writes it. See the database-migration skill for DDL hygiene.

How to test: This is a structural/review check. In CI, assert that application (non-migration) modules emit no DDL — e.g. grep for CREATE TABLE/ALTER TABLE outside the alembic/versions path, or fail the test suite if the engine logs DDL during a request-path test.

---

6. Parameterize queries — never interpolate untrusted data

Why: String-building SQL with user input is the classic SQL-injection vector and also defeats the database's prepared-statement plan cache. Always bind parameters. SQLAlchemy's expression language and text() with bound params do this for you; raw f-strings do not.

Non-compliant:

session.execute(text(f"SELECT * FROM users WHERE email = '{email}'"))  # injectable

Compliant:

session.execute(
    text("SELECT id FROM users WHERE email = :email"), {"email": email}
)
# or, preferred, the expression language:
session.scalar(select(User.id).where(User.email == email))

How to test: Add a test that passes an injection payload (e.g. x'; DROP TABLE …) and asserts it is treated as a literal value (no rows / no side effect), plus a lint/grep gate forbidding f-string/%/+ SQL construction.

---

7. Prefer independent subqueries (or joins) over correlated subqueries

Why: A correlated subquery re-executes per outer row; an independent subquery (or a plain join with proper indexes) runs once. Complex nested subqueries also hurt readability and changeability.

Non-compliant — correlated, runs per outer row:

SELECT o.id
FROM orders o
WHERE o.amount > (SELECT AVG(amount) FROM orders WHERE customer_id = o.customer_id);

Compliant — compute the aggregate once, then join:

WITH avg_per_customer AS (
    SELECT customer_id, AVG(amount) AS avg_amount FROM orders GROUP BY customer_id
)
SELECT o.id
FROM orders o
JOIN avg_per_customer a ON a.customer_id = o.customer_id
WHERE o.amount > a.avg_amount;

How to test: Compare EXPLAIN cost/loops between the two forms on representative data, or assert result-set equality between the correlated and join/CTE rewrite to prove the refactor is behavior-preserving.

---

Where this fits

These are query-design checks complementary to the ORM mechanics in the main skill. For DDL hygiene and migration ordering see the database-migration skill; for the project-wide efficiency checklist see code-review-standards (criteria-efficiency.md).