Legacy 1.4 to 2.0 Codemod Checklist

Migrate from SQLAlchemy 1.4 to 2.0 by enabling future=True, resolving all RemovedIn20Warning emissions, then systematically replacing session.query(), Query.get(), autocommit, backref, declarative_base(), and Column() with their 2.0 equivalents before removing the transitional flag.

This checklist belongs to the Migrating Legacy 1.4 Code to 2.0 Syntax guide. Work through the steps in the order shown — they build on each other, and skipping ahead will produce confusing errors.

Quick Answer

The highest-impact single change is replacing session.query() with select() plus session.execute(). Every other step is important, but this one touches the most lines of application code.

Legacy 1.4 pattern:

from sqlalchemy.orm import Session
from myapp.models import User

# Synchronous session.query() — removed in 2.0
def get_active_users(session: Session) -> list[User]:
    return (
        session.query(User)
        .filter(User.is_active == True)
        .order_by(User.created_at.desc())
        .all()
    )

# Query.get() — also removed in 2.0
def get_user_by_pk(session: Session, user_id: int) -> User | None:
    return session.query(User).get(user_id)

Modern 2.0 pattern:

from sqlalchemy import select
from sqlalchemy.orm import Session
from myapp.models import User

# select() + session.execute() — the 2.0 way
def get_active_users(session: Session) -> list[User]:
    stmt = (
        select(User)
        .where(User.is_active == True)
        .order_by(User.created_at.desc())
    )
    return session.execute(stmt).scalars().all()

# Session.get() replaces Query.get()
def get_user_by_pk(session: Session, user_id: int) -> User | None:
    return session.get(User, user_id)

Execution Context & Async Workflow Integration

Every codemod step has a direct consequence for async code. Working through them in order ensures you do not introduce async-incompatible patterns while you are mid-migration.

Step 1 — Enable future=True on create_engine()

future=True activates 2.0-style behavior inside a 1.4 install. This is the migration bridge.

from sqlalchemy import create_engine
from sqlalchemy.ext.asyncio import create_async_engine

# Synchronous engine — add future=True
engine = create_engine(
    "postgresql+psycopg2://user:pass@localhost/mydb",
    future=True,
)

# Async engine — already implies future=True, but explicit is fine
async_engine = create_async_engine(
    "postgresql+asyncpg://user:pass@localhost/mydb",
    future=True,
)

With future=True set, the engine enforces 2.0 connection semantics: autobegin, no autocommit, and Core text queries require text(). This is the minimum change that makes subsequent steps testable.

Step 2 — Set SQLALCHEMY_WARN_20=1 and audit all warnings

Before you change any model or query code, turn on every deprecation warning. Run your test suite with the environment variable set and capture the output.

SQLALCHEMY_WARN_20=1 python -W error::sqlalchemy.exc.RemovedIn20Warning -m pytest tests/

The -W error flag promotes every RemovedIn20Warning to a hard exception, making tests fail at the exact call site instead of printing a warning you can ignore. Fix warnings in the order your test output shows them — that order usually tracks call frequency, so the busiest code paths surface first.

Step 3 — Replace session.query() with select() + session.execute()

This is the largest mechanical change. The async implication is significant: AsyncSession never supported session.query() at all, so any code that ran synchronously via session.query() cannot be directly ported to async without this step.

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from myapp.models import Order, User

async def get_open_orders_for_user(
    session: AsyncSession,
    user_id: int,
) -> list[Order]:
    stmt = (
        select(Order)
        .where(Order.user_id == user_id, Order.status == "open")
        .order_by(Order.created_at.desc())
    )
    result = await session.execute(stmt)
    return result.scalars().all()

The .scalars() call is required whenever you select a single ORM entity. Without it, session.execute() returns Row objects rather than mapped instances. For multi-entity queries, drop .scalars() and unpack the Row tuples directly.

Step 4 — Replace Query.get(pk) with Session.get(Model, pk)

Session.get() and AsyncSession.get() hit the identity map first, then issue a SELECT only on a cache miss — the same semantics as Query.get().

from sqlalchemy.ext.asyncio import AsyncSession
from myapp.models import Invoice

async def fetch_invoice(session: AsyncSession, invoice_id: int) -> Invoice | None:
    # Session.get() works identically in sync and async contexts
    return await session.get(Invoice, invoice_id)

Composite primary keys are supported as a tuple: session.get(Order, (tenant_id, order_id)).

Step 5 — Remove autocommit=True — switch to explicit session.begin() / commit()

SQLAlchemy 2.0 removed the autocommit Session flag. The replacement is explicit transaction management. In async code this matters especially because you often scope a session to a single request or task.

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

engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/mydb")
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)

async def create_tenant(name: str) -> None:
    async with AsyncSessionLocal() as session:
        async with session.begin():          # explicit transaction
            tenant = Tenant(name=name)
            session.add(tenant)
        # commit fires automatically when the inner context exits cleanly

The async with session.begin() context manager commits on success and rolls back on exception. Never rely on implicit commit in 2.0. For more on commit strategy patterns, see Transaction Isolation and Commit Strategies.

Step 6 — Update relationship() to use back_populates instead of backref

backref still works in 2.0 but is considered a legacy shortcut. back_populates makes both sides of the relationship explicit, which is required when you add Mapped[T] annotations in the next step.

# Legacy — backref implicitly creates the reverse attribute
class User(Base):
    orders = relationship("Order", backref="user")

# 2.0 — both sides declared, compatible with Mapped[T]
class User(Base):
    orders: Mapped[list["Order"]] = relationship(back_populates="user")

class Order(Base):
    user: Mapped["User"] = relationship(back_populates="orders")

Step 7 — Replace declarative_base() with a DeclarativeBase subclass

The module-level declarative_base() factory is replaced by subclassing DeclarativeBase. This unlocks PEP 681 dataclass integration and proper type inference.

# Legacy
from sqlalchemy.orm import declarative_base
Base = declarative_base()

# 2.0
from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
    pass

Your existing model classes continue to inherit from Base without any other changes. The metadata and registry attributes are now class-level attributes on your Base subclass rather than module globals.

Step 8 — Replace Column() with mapped_column() + Mapped[T] annotations

This step is the deepest change and the one that delivers the most long-term value: full static type-checking for ORM models. See the Step-by-Step Guide to SQLAlchemy 2.0 Type Annotations for the full annotation migration walkthrough.

from __future__ import annotations

from datetime import datetime
from decimal import Decimal

from sqlalchemy import Numeric, String, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

class Base(DeclarativeBase):
    pass

class Product(Base):
    __tablename__ = "products"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(255), nullable=False)
    price: Mapped[Decimal] = mapped_column(Numeric(10, 2))
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())
    description: Mapped[str | None] = mapped_column(String(2000))

The Mapped[T] annotation drives nullability: Mapped[str] implies NOT NULL, while Mapped[str | None] implies nullable=True. You can drop explicit nullable= arguments once annotations are in place.

Step 9 — Remove future=True and unset SQLALCHEMY_WARN_20

When your test suite passes with zero RemovedIn20Warning emissions and all model code uses Mapped[T] annotations, remove the transitional future=True argument (it is the default and has no effect in 2.0) and stop setting SQLALCHEMY_WARN_20.

# Clean 2.0 engine — no future= needed
engine = create_async_engine(
    "postgresql+asyncpg://user:pass@localhost/mydb",
    pool_size=10,
    max_overflow=20,
)

At this point your codebase is fully 2.0-native. You can now safely upgrade from SQLAlchemy 1.4 to the latest 2.x release.

Resolving Warnings, Errors & Common Mistakes

Advanced Codemod Optimization

Turn the warning stream itself into your worklist instead of hand-grepping for call sites. SQLAlchemy does not ship an official AST codemod, and the abandoned sqlalchemy-migrate package on PyPI is unrelated — it is a 0.x-era schema tool, not a 1.4→2.0 rewriter. The reliable mechanical step is to let the 1.4 RemovedIn20Warning machinery enumerate every offending line for you, because each warning carries the file, line number, and the exact recommended replacement in its message text.

Run your full test suite once with warnings captured to a file, then bucket the results by category so each codemod step becomes a finite, reviewable checklist:

# capture_warnings.py — run under SQLALCHEMY_WARN_20=1
import warnings
from collections import Counter

records: list[warnings.WarningMessage] = []
orig_show = warnings.showwarning


def record(message, category, filename, lineno, file=None, line=None):
    records.append(
        warnings.WarningMessage(message, category, filename, lineno, file, line)
    )
    orig_show(message, category, filename, lineno, file, line)


warnings.showwarning = record  # install before importing your app

# ... import and exercise your app / run pytest in-process ...

by_site = Counter((r.filename, r.lineno) for r in records)
for (path, line), count in by_site.most_common():
    print(f"{path}:{line}  x{count}  {records[0].message}")

This produces a deduplicated, frequency-ranked list of exact path:line sites — far more precise than grep -rn "session\.query", which misses dynamically built queries and reports false positives inside strings and comments. Commit each category as its own atomic changeset (one for session.query() → select(), one for Query.get() → Session.get(), and so on) so reviewers can separate mechanical renames from the judgment-intensive steps such as transaction strategy and Mapped[T] annotations. Pair this with grep -rn "session\.query" only as a final sweep to catch call sites your tests never execute.

Frequently Asked Questions

Can I migrate one module at a time, or must I update the whole codebase at once? You can migrate incrementally. With future=True on the engine and SQLALCHEMY_WARN_20=1 set, legacy and 2.0-style code coexist in the same process. Migrate module by module, running your test suite after each batch. The key constraint is that all code sharing a single Session instance should use consistent patterns — mixing session.query() and select() inside one unit of work is safe but confusing during review.

Do I need to update Alembic as well? Yes. Alembic 1.8+ supports SQLAlchemy 2.0. Upgrade Alembic before you upgrade SQLAlchemy: run pip install --upgrade alembic sqlalchemy, then verify your migration scripts still generate correctly with alembic revision --autogenerate. Alembic autogenerate reads Base.metadata, which is unaffected by most codemod steps, so migration history is preserved.

What happens to expire_on_commit when I switch to session.begin()? Nothing changes by default — expire_on_commit=True is still the default, meaning all attributes are expired after a commit. In async code, accessing an expired attribute after the session closes raises MissingGreenlet. The recommended fix is async_sessionmaker(engine, expire_on_commit=False) combined with explicit re-fetching when you need fresh data, rather than relying on lazy reload.

How do I handle backref in third-party models I cannot edit? If you inherit from a third-party base that still uses backref, you cannot remove it without forking the library. Add the relationship on your own subclass using back_populates, and file an upstream issue requesting the migration. SQLAlchemy 2.0 does not break backref — it only deprecates it — so you have time to wait for the upstream fix without blocking your own migration. For the full picture of what Core vs ORM architecture means for third-party integration points, see the architecture decisions page.

Related

• Step-by-Step Guide to SQLAlchemy 2.0 Type Annotations — deep dive into Mapped[T] and mapped_column() after completing this checklist.
• Fixing RemovedIn20Warning Deprecation Warnings — targeted remediation for each warning category the audit step surfaces.
• Session Lifecycle and Scope Management — context for the explicit transaction patterns introduced in Step 5.
• Transaction Isolation and Commit Strategies — production patterns for session.begin() and rollback handling across sync and async code.