sqlalchemy 2.0.44: Move a SQLAlchemy 1.4 query to the 2.0 select API and get entities back instead of Row tuples

{"case":{"caseId":"case:sha256:b0c05249719b4c637f99ca3e4886db933b3a88cf9ecdaffb3c658060f43263ea","constraints":{"runtime":"python"},"contract":["assert session.execute(select(...)) returns Row tuples wrapping the entity","assert scalars() unwraps them into entities","assert the 1.x Query API still works and query().get() only warns","assert Engine.execute is gone and a bare SQL string is not executable","assert Connection and Session raise different exceptions for the same bare string"],"goal":"Move a SQLAlchemy 1.4 query to the 2.0 select API and get entities back instead of Row tuples","kind":"MIGRATION","packages":["pkg:pypi/sqlalchemy@2.0.44"],"schemaVersion":1,"symbols":["Session.execute","Session.scalars","select","sqlalchemy.text","DeclarativeBase"]},"contractCommand":["python","test/contract.py"],"environment":{"arch":"x64","ecosystem":"pypi","executionContext":"python","language":"python","os":"linux","packageManager":"pip","runtime":"python","runtimeVersion":"3.12","schemaVersion":1},"license":"MIT-0","packages":["pkg:pypi/sqlalchemy@2.0.44"],"schemaVersion":1,"symbols":["Session.execute","Session.scalars","select","sqlalchemy.text","DeclarativeBase"],"verifierAdapter":"pypi@1"}

sqlalchemy==2.0.44
typing-extensions==4.16.0
greenlet==3.2.4

"""SQLAlchemy 1.4 -> 2.0, in the two places it actually costs time.

The first is quiet. session.execute(select(User)) does not give you User
objects — it gives you Row tuples with a User inside each one, exactly like
a SQL driver would. Code written against Query gets a list of one-element
tuples, and the failure surfaces later as an attribute error on a tuple.
.scalars() is the unwrapping step, and it is the whole difference between
the two idioms.

The second is loud but easy to misread. A bare SQL string is no longer
executable, and the exception depends on where you passed it: a Connection
says ObjectNotExecutableError, a Session says ArgumentError. Both mean
wrap it in text().

What did NOT change, measured rather than assumed: Query still works, and
session.query(User).get(1) warns instead of failing. So a 2.0 upgrade can
run green while none of the code is 2.0 yet.
"""

import sqlalchemy as sa
from sqlalchemy import select
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column


class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]


def seeded_engine() -> sa.Engine:
    engine = sa.create_engine("sqlite+pysqlite:///:memory:")
    Base.metadata.create_all(engine)
    with Session(engine) as session:
        session.add_all([User(id=1, name="ada"), User(id=2, name="linus")])
        session.commit()
    return engine


def rows(session: Session) -> list:
    """The 2.0 call that surprises people: Row tuples, not entities."""
    return session.execute(select(User)).all()


def entities(session: Session) -> list[User]:
    """The same query, unwrapped. This is what Query used to return."""
    return session.scalars(select(User)).all()

import sys
import warnings
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

import sqlalchemy as sa
from sqlalchemy import select
from sqlalchemy.orm import Session

from src.store import User, entities, rows, seeded_engine

engine = seeded_engine()

with Session(engine) as session:
    # execute() returns Row tuples that WRAP the entity.
    first = rows(session)[0]
    assert type(first).__name__ == "Row", type(first)
    assert isinstance(first[0], User), first
    assert not isinstance(first, User)

    # scalars() unwraps them. This is the line most migrations are missing.
    people = entities(session)
    assert [u.name for u in people] == ["ada", "linus"], people

    # The 1.x Query API was not removed, which is why an upgrade can look
    # finished before it has started.
    assert [u.name for u in session.query(User).all()] == ["ada", "linus"]

    with warnings.catch_warnings(record=True) as caught:
        warnings.simplefilter("always")
        session.query(User).get(1)
    assert [type(c.message).__name__ for c in caught] == ["LegacyAPIWarning"], caught

    # session.get is the replacement, and it returns the entity directly.
    assert session.get(User, 1).name == "ada"
    assert session.scalars(select(User).where(User.id == 99)).one_or_none() is None

# Connectionless execution really is gone: this one is an AttributeError,
# not a deprecation.
assert not hasattr(engine, "execute")

# A bare SQL string is not executable, and the two entry points disagree
# about which exception says so.
with engine.connect() as conn:
    try:
        conn.execute("select 1")
        raise AssertionError("a bare string should not be executable")
    except sa.exc.ObjectNotExecutableError:
        pass
    assert list(conn.execute(sa.text("select 1"))) == [(1,)]

with Session(engine) as session:
    try:
        session.execute("select 1")
        raise AssertionError("a bare string should not be executable")
    except sa.exc.ArgumentError as exc:
        assert "text(" in str(exc), exc

print("contract ok:", sa.__version__)