sqlalchemy-consistency

SQLAlchemy — consistent, modern idioms

SQLAlchemy is stable and extremely well represented in training data, which is exactly the problem: generated code mixes three eras — pre-1.4 patterns (implicit autocommit, bare Column with declarative_base()), the 1.x Query API (session.query(User).filter(...)), and modern 2.0 style (select() + session.execute(), Mapped[...] typed models). This skill pins one canonical idiom set — SQLAlchemy 2.0 style only — so every snippet you produce or review follows the same rules instead of mixing eras.

Canonical idioms — always X, never Y

Always	Never	Why
select(User).where(User.name == n) + session.scalars(stmt)	session.query(User).filter(...)	Query is the legacy 1.x API; 2.0 unifies Core and ORM on select(). Never mix both in one file.
class Base(DeclarativeBase): pass + Mapped[int] / mapped_column()	Base = declarative_base() + bare Column(Integer)	Typed declarative is the 2.0 mapping style; it gives type checkers and IDEs real types.
with Session(engine) as session, session.begin():	relying on autocommit or leaving sessions open	Pre-1.4 implicit autocommit is removed; uncommitted work is silently rolled back at GC.
text("... WHERE id = :id") + {"id": x} params	f-string / %-formatted SQL strings	String interpolation is an injection hole and skips type binding. Also: bare strings to execute() were removed — wrap in text().
session.scalars(stmt).all() for entities	session.execute(stmt).all() then row[0] everywhere	execute returns Row tuples; scalars unwraps the single entity. Picking the wrong one is the top row-shape confusion.
relationship(back_populates=...) typed on both sides	relationship(backref=...)	backref is legacy/discouraged; back_populates is explicit and type-checkable.
selectinload(User.posts) / joinedload options up front	lazy-loading in loops or after session.close()	Lazy loads cause N+1 and raise DetachedInstanceError once the session is gone.
engine.begin() for write blocks, engine.connect() for reads	engine.execute(...)	Connectionless/implicit execution was removed in 2.0.
create_engine(url) once per process, reuse it	create_engine inside request handlers	The engine owns the pool; recreating it leaks connections and defeats pooling.
session.get(User, pk)	session.query(User).get(pk)	Query.get() is legacy; Session.get() is the 2.0 spelling.
stmt = stmt.where(...) (reassign)	mutating in place / expecting it	Statements are immutable generatively; the un-reassigned where() is a silent no-op.

House style for a model and a query:

from sqlalchemy import ForeignKey, String, create_engine, select
from sqlalchemy.orm import (DeclarativeBase, Mapped, Session,
                            mapped_column, relationship, selectinload)

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(255), unique=True)
    posts: Mapped[list["Post"]] = relationship(back_populates="author")

class Post(Base):
    __tablename__ = "posts"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str]
    author_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
    author: Mapped[User] = relationship(back_populates="posts")

engine = create_engine("postgresql+psycopg://app@localhost/app")

with Session(engine) as session, session.begin():
    stmt = (
        select(User)
        .where(User.email.like("%@example.com"))
        .options(selectinload(User.posts))
        .order_by(User.id)
    )
    users = session.scalars(stmt).all()

Pitfalls that produce silently wrong results

• Forgetting commit: with no autocommit, a session that exits without commit() rolls back. Use the session.begin() block so commit/rollback is structural, not remembered.
• Row vs entity shape: session.execute(select(User)).all() yields Row objects — row.User.email, not row.email. Code written for scalars() but run through execute() (or vice versa) breaks subtly or serializes tuples to JSON.
• .where() without reassignment builds a new statement and discards it; the query silently returns unfiltered rows.
• joinedload against a collection + LIMIT multiplies/clips rows unless you add .unique() on the scalars result (required, or it raises) — prefer selectinload for collections, joinedload for many-to-one.
• DetachedInstanceError after close: load everything you need (eager-load options or access inside the session block) before the session ends; don't widen object lifetimes to dodge it.
• Session shared across threads/tasks: sessions are not thread-safe; one session per thread/request/task, engine shared globally. In async code, never let an AsyncSession lazy-load implicitly — that's the MissingGreenlet error; eager-load or await session.refresh(obj, ["rel"]).
• filter_by(name=x) vs where(User.name == x): filter_by exists in 2.0 but only keyword-matches the primary entity — house style is where() with explicit column expressions everywhere.
• In-place mutation of JSON/array columns isn't change-tracked by default; reassign the attribute or use MutableDict.as_mutable(...).

Version notes

Target SQLAlchemy 2.0. The breaking line is 1.4 → 2.0: implicit autocommit, bare-string execute(), and connectionless execution are removed; Query survives as legacy but must not be mixed with select() style. Code written 2.0-style runs on 1.4 (the migration bridge release) — so write 2.0 style unconditionally; only note differences if the user is pinned below 1.4. Async needs create_async_engine + AsyncSession and an async driver (postgresql+asyncpg); the statement API is identical, only execution is awaited. Schema migrations are Alembic's job — out of scope here; Base.metadata.create_all(engine) is fine for tests and throwaway scripts only.

Workflow

1. Models first: DeclarativeBase subclass, Mapped[...] annotations, mapped_column() for constraints, relationships typed with back_populates on both sides.
2. One engine per process; sessions are short-lived: with Session(engine) as session, session.begin(): (or a sessionmaker factory wired the same way).
3. Build queries with select()/insert()/update()/delete(); reassign on every generative call; execute via session.scalars() for entities, session.execute() for multi-column rows.
4. Decide loading strategy at query time: selectinload for collections, joinedload for many-to-one; never rely on lazy loads outside the session.
5. Raw SQL only through text() with :name bound parameters.
6. When reviewing existing code, flag any "Never" column pattern above — especially session.query mixed with select() — and rewrite in canonical form rather than patching around it.

For the fuller migration map (1.x → 2.0), async specifics, result-shape tables, and more worked examples, read references/sqlalchemy-patterns.md.