Merge "Convert remaining ORM APIs to support 2.0 style"

16 files changed, 1699 insertions, 406 deletions

diff --git a/test/aaa_profiling/test_memusage.py b/test/aaa_profiling/test_memusage.py
index f141dbcc9..5e388c0b7 100644
--- a/test/aaa_profiling/test_memusage.py
+++ b/test/aaa_profiling/test_memusage.py
@@ -402,31 +402,30 @@ class MemUsageWBackendTest(EnsureZeroed):
 
         @profile_memory()
         def go():
-            sess = create_session()
-            a1 = A(col2="a1")
-            a2 = A(col2="a2")
-            a3 = A(col2="a3")
-            a1.bs.append(B(col2="b1"))
-            a1.bs.append(B(col2="b2"))
-            a3.bs.append(B(col2="b3"))
-            for x in [a1, a2, a3]:
-                sess.add(x)
-            sess.flush()
-            sess.expunge_all()
-
-            alist = sess.query(A).order_by(A.col1).all()
-            eq_(
-                [
-                    A(col2="a1", bs=[B(col2="b1"), B(col2="b2")]),
-                    A(col2="a2", bs=[]),
-                    A(col2="a3", bs=[B(col2="b3")]),
-                ],
-                alist,
-            )
+            with Session() as sess:
+                a1 = A(col2="a1")
+                a2 = A(col2="a2")
+                a3 = A(col2="a3")
+                a1.bs.append(B(col2="b1"))
+                a1.bs.append(B(col2="b2"))
+                a3.bs.append(B(col2="b3"))
+                for x in [a1, a2, a3]:
+                    sess.add(x)
+                sess.commit()
+
+                alist = sess.query(A).order_by(A.col1).all()
+                eq_(
+                    [
+                        A(col2="a1", bs=[B(col2="b1"), B(col2="b2")]),
+                        A(col2="a2", bs=[]),
+                        A(col2="a3", bs=[B(col2="b3")]),
+                    ],
+                    alist,
+                )
 
-            for a in alist:
-                sess.delete(a)
-            sess.flush()
+                for a in alist:
+                    sess.delete(a)
+                sess.commit()
 
         go()
 
@@ -501,33 +500,31 @@ class MemUsageWBackendTest(EnsureZeroed):
                     "use_reaper": False,
                 }
             )
-            sess = create_session(bind=engine)
-
-            a1 = A(col2="a1")
-            a2 = A(col2="a2")
-            a3 = A(col2="a3")
-            a1.bs.append(B(col2="b1"))
-            a1.bs.append(B(col2="b2"))
-            a3.bs.append(B(col2="b3"))
-            for x in [a1, a2, a3]:
-                sess.add(x)
-            sess.flush()
-            sess.expunge_all()
+            with Session(engine) as sess:
+                a1 = A(col2="a1")
+                a2 = A(col2="a2")
+                a3 = A(col2="a3")
+                a1.bs.append(B(col2="b1"))
+                a1.bs.append(B(col2="b2"))
+                a3.bs.append(B(col2="b3"))
+                for x in [a1, a2, a3]:
+                    sess.add(x)
+                sess.commit()
+
+                alist = sess.query(A).order_by(A.col1).all()
+                eq_(
+                    [
+                        A(col2="a1", bs=[B(col2="b1"), B(col2="b2")]),
+                        A(col2="a2", bs=[]),
+                        A(col2="a3", bs=[B(col2="b3")]),
+                    ],
+                    alist,
+                )
 
-            alist = sess.query(A).order_by(A.col1).all()
-            eq_(
-                [
-                    A(col2="a1", bs=[B(col2="b1"), B(col2="b2")]),
-                    A(col2="a2", bs=[]),
-                    A(col2="a3", bs=[B(col2="b3")]),
-                ],
-                alist,
-            )
+                for a in alist:
+                    sess.delete(a)
+                sess.commit()
 
-            for a in alist:
-                sess.delete(a)
-            sess.flush()
-            sess.close()
             engine.dispose()
 
         go()
@@ -555,29 +552,27 @@ class MemUsageWBackendTest(EnsureZeroed):
         mapper(Wide, wide_table, _compiled_cache_size=10)
 
         metadata.create_all()
-        session = create_session()
-        w1 = Wide()
-        session.add(w1)
-        session.flush()
-        session.close()
+        with Session() as session:
+            w1 = Wide()
+            session.add(w1)
+            session.commit()
         del session
         counter = [1]
 
         @profile_memory()
         def go():
-            session = create_session()
-            w1 = session.query(Wide).first()
-            x = counter[0]
-            dec = 10
-            while dec > 0:
-                # trying to count in binary here,
-                # works enough to trip the test case
-                if pow(2, dec) < x:
-                    setattr(w1, "col%d" % dec, counter[0])
-                    x -= pow(2, dec)
-                dec -= 1
-            session.flush()
-            session.close()
+            with Session() as session:
+                w1 = session.query(Wide).first()
+                x = counter[0]
+                dec = 10
+                while dec > 0:
+                    # trying to count in binary here,
+                    # works enough to trip the test case
+                    if pow(2, dec) < x:
+                        setattr(w1, "col%d" % dec, counter[0])
+                        x -= pow(2, dec)
+                    dec -= 1
+                session.commit()
             counter[0] += 1
 
         try:
diff --git a/test/orm/inheritance/test_basic.py b/test/orm/inheritance/test_basic.py
index 589ef3f52..5d09d5e58 100644
--- a/test/orm/inheritance/test_basic.py
+++ b/test/orm/inheritance/test_basic.py
@@ -1913,33 +1913,32 @@ class VersioningTest(fixtures.MappedTest):
         )
         mapper(Sub, subtable, inherits=Base, polymorphic_identity=2)
 
-        sess = create_session()
+        sess = Session(autoflush=False)
 
         b1 = Base(value="b1")
         s1 = Sub(value="sub1", subdata="some subdata")
         sess.add(b1)
         sess.add(s1)
 
-        sess.flush()
+        sess.commit()
 
-        sess2 = create_session()
-        s2 = sess2.query(Base).get(s1.id)
+        sess2 = Session(autoflush=False)
+        s2 = sess2.get(Base, s1.id)
         s2.subdata = "sess2 subdata"
 
         s1.subdata = "sess1 subdata"
 
-        sess.flush()
+        sess.commit()
 
         assert_raises(
             orm_exc.StaleDataError,
-            sess2.query(Base).with_for_update(read=True).get,
+            sess2.get,
+            Base,
             s1.id,
+            with_for_update=dict(read=True),
         )
 
-        if not testing.db.dialect.supports_sane_rowcount:
-            sess2.flush()
-        else:
-            assert_raises(orm_exc.StaleDataError, sess2.flush)
+        sess2.rollback()
 
         sess2.refresh(s2)
         if testing.db.dialect.supports_sane_rowcount:
@@ -1967,7 +1966,7 @@ class VersioningTest(fixtures.MappedTest):
         )
         mapper(Sub, subtable, inherits=Base, polymorphic_identity=2)
 
-        sess = create_session()
+        sess = Session(autoflush=False, expire_on_commit=False)
 
         b1 = Base(value="b1")
         s1 = Sub(value="sub1", subdata="some subdata")
@@ -1976,21 +1975,21 @@ class VersioningTest(fixtures.MappedTest):
         sess.add(s1)
         sess.add(s2)
 
-        sess.flush()
+        sess.commit()
 
-        sess2 = create_session()
-        s3 = sess2.query(Base).get(s1.id)
+        sess2 = Session(autoflush=False, expire_on_commit=False)
+        s3 = sess2.get(Base, s1.id)
         sess2.delete(s3)
-        sess2.flush()
+        sess2.commit()
 
         s2.subdata = "some new subdata"
-        sess.flush()
+        sess.commit()
 
         s1.subdata = "some new subdata"
         if testing.db.dialect.supports_sane_rowcount:
-            assert_raises(orm_exc.StaleDataError, sess.flush)
+            assert_raises(orm_exc.StaleDataError, sess.commit)
         else:
-            sess.flush()
+            sess.commit()
 
 
 class DistinctPKTest(fixtures.MappedTest):
diff --git a/test/orm/test_deprecations.py b/test/orm/test_deprecations.py
index 7689663b1..43ab3d630 100644
--- a/test/orm/test_deprecations.py
+++ b/test/orm/test_deprecations.py
@@ -3,6 +3,7 @@ from sqlalchemy import and_
 from sqlalchemy import cast
 from sqlalchemy import desc
 from sqlalchemy import event
+from sqlalchemy import exc as sa_exc
 from sqlalchemy import func
 from sqlalchemy import Integer
 from sqlalchemy import literal_column
@@ -35,6 +36,8 @@ from sqlalchemy.orm import undefer
 from sqlalchemy.orm import with_polymorphic
 from sqlalchemy.orm.collections import collection
 from sqlalchemy.orm.util import polymorphic_union
+from sqlalchemy.sql import elements
+from sqlalchemy.testing import assert_raises
 from sqlalchemy.testing import assert_raises_message
 from sqlalchemy.testing import assertions
 from sqlalchemy.testing import AssertsCompiledSQL
@@ -51,6 +54,7 @@ from .inheritance import _poly_fixtures
 from .test_events import _RemoveListeners
 from .test_options import PathTest as OptionsPathTest
 from .test_query import QueryTest
+from .test_transaction import _LocalFixture
 
 
 class DeprecatedQueryTest(_fixtures.FixtureTest, AssertsCompiledSQL):
@@ -510,6 +514,127 @@ class DeprecatedQueryTest(_fixtures.FixtureTest, AssertsCompiledSQL):
             )
 
 
+class SessionTest(fixtures.RemovesEvents, _LocalFixture):
+    def test_subtransactions_deprecated(self):
+        s1 = Session(testing.db)
+        s1.begin()
+
+        with testing.expect_deprecated_20(
+            "The Session.begin.subtransactions flag is deprecated "
+            "and will be removed in SQLAlchemy version 2.0."
+        ):
+            s1.begin(subtransactions=True)
+
+        s1.close()
+
+    def test_autocommit_deprecated(Self):
+        with testing.expect_deprecated_20(
+            "The Session.autocommit parameter is deprecated "
+            "and will be removed in SQLAlchemy version 2.0."
+        ):
+            Session(autocommit=True)
+
+    @testing.requires.independent_connections
+    @testing.emits_warning(".*previous exception")
+    def test_failed_rollback_deactivates_transaction_ctx_integration(self):
+        # test #4050 in the same context as that of oslo.db
+
+        User = self.classes.User
+
+        with testing.expect_deprecated_20(
+            "The Session.autocommit parameter is deprecated"
+        ):
+            session = Session(bind=testing.db, autocommit=True)
+
+        evented_exceptions = []
+        caught_exceptions = []
+
+        def canary(context):
+            evented_exceptions.append(context.original_exception)
+
+        rollback_error = testing.db.dialect.dbapi.InterfaceError(
+            "Can't roll back to savepoint"
+        )
+
+        def prevent_savepoint_rollback(
+            cursor, statement, parameters, context=None
+        ):
+            if (
+                context is not None
+                and context.compiled
+                and isinstance(
+                    context.compiled.statement,
+                    elements.RollbackToSavepointClause,
+                )
+            ):
+                raise rollback_error
+
+        self.event_listen(testing.db, "handle_error", canary, retval=True)
+        self.event_listen(
+            testing.db.dialect, "do_execute", prevent_savepoint_rollback
+        )
+
+        with session.begin():
+            session.add(User(id=1, name="x"))
+
+        try:
+            with session.begin():
+                try:
+                    with session.begin_nested():
+                        # raises IntegrityError on flush
+                        session.add(User(id=1, name="x"))
+
+                # outermost is the failed SAVEPOINT rollback
+                # from the "with session.begin_nested()"
+                except sa_exc.DBAPIError as dbe_inner:
+                    caught_exceptions.append(dbe_inner.orig)
+                    raise
+        except sa_exc.DBAPIError as dbe_outer:
+            caught_exceptions.append(dbe_outer.orig)
+
+        is_true(
+            isinstance(
+                evented_exceptions[0], testing.db.dialect.dbapi.IntegrityError
+            )
+        )
+        eq_(evented_exceptions[1], rollback_error)
+        eq_(len(evented_exceptions), 2)
+        eq_(caught_exceptions, [rollback_error, rollback_error])
+
+    def test_contextmanager_commit(self):
+        User = self.classes.User
+
+        with testing.expect_deprecated_20(
+            "The Session.autocommit parameter is deprecated"
+        ):
+            sess = Session(autocommit=True)
+        with sess.begin():
+            sess.add(User(name="u1"))
+
+        sess.rollback()
+        eq_(sess.query(User).count(), 1)
+
+    def test_contextmanager_rollback(self):
+        User = self.classes.User
+
+        with testing.expect_deprecated_20(
+            "The Session.autocommit parameter is deprecated"
+        ):
+            sess = Session(autocommit=True)
+
+        def go():
+            with sess.begin():
+                sess.add(User())  # name can't be null
+
+        assert_raises(sa_exc.DBAPIError, go)
+
+        eq_(sess.query(User).count(), 0)
+
+        with sess.begin():
+            sess.add(User(name="u1"))
+        eq_(sess.query(User).count(), 1)
+
+
 class DeprecatedInhTest(_poly_fixtures._Polymorphic):
     def test_with_polymorphic(self):
         Person = _poly_fixtures.Person
diff --git a/test/orm/test_dynamic.py b/test/orm/test_dynamic.py
index 753a97ae1..27883a5de 100644
--- a/test/orm/test_dynamic.py
+++ b/test/orm/test_dynamic.py
@@ -12,10 +12,8 @@ from sqlalchemy.orm import configure_mappers
 from sqlalchemy.orm import create_session
 from sqlalchemy.orm import exc as orm_exc
 from sqlalchemy.orm import mapper
-from sqlalchemy.orm import Query
 from sqlalchemy.orm import relationship
 from sqlalchemy.orm import Session
-from sqlalchemy.orm.dynamic import AppenderMixin
 from sqlalchemy.testing import assert_raises
 from sqlalchemy.testing import assert_raises_message
 from sqlalchemy.testing import AssertsCompiledSQL
@@ -67,6 +65,62 @@ class _DynamicFixture(object):
         mapper(Item, items)
         return Order, Item
 
+    def _user_order_item_fixture(self):
+        (
+            users,
+            Keyword,
+            items,
+            order_items,
+            item_keywords,
+            Item,
+            User,
+            keywords,
+            Order,
+            orders,
+        ) = (
+            self.tables.users,
+            self.classes.Keyword,
+            self.tables.items,
+            self.tables.order_items,
+            self.tables.item_keywords,
+            self.classes.Item,
+            self.classes.User,
+            self.tables.keywords,
+            self.classes.Order,
+            self.tables.orders,
+        )
+
+        mapper(
+            User,
+            users,
+            properties={
+                "orders": relationship(
+                    Order, order_by=orders.c.id, lazy="dynamic"
+                )
+            },
+        )
+        mapper(
+            Order,
+            orders,
+            properties={
+                "items": relationship(
+                    Item, secondary=order_items, order_by=items.c.id
+                ),
+            },
+        )
+        mapper(
+            Item,
+            items,
+            properties={
+                "keywords": relationship(
+                    Keyword, secondary=item_keywords
+                )  # m2m
+            },
+        )
+        mapper(Keyword, keywords)
+
+        return User, Order, Item, Keyword
+
 
 class DynamicTest(_DynamicFixture, _fixtures.FixtureTest, AssertsCompiledSQL):
     def test_basic(self):
@@ -117,11 +171,10 @@ class DynamicTest(_DynamicFixture, _fixtures.FixtureTest, AssertsCompiledSQL):
         sess = create_session()
         u = sess.query(User).get(8)
         sess.expunge(u)
-        assert_raises(
-            orm_exc.DetachedInstanceError,
-            u.addresses.filter_by,
-            email_address="e",
-        )
+
+        q = u.addresses.filter_by(email_address="e")
+
+        assert_raises(orm_exc.DetachedInstanceError, q.first)
 
     def test_no_uselist_false(self):
         User, Address = self._user_address_fixture(
@@ -450,6 +503,12 @@ class DynamicTest(_DynamicFixture, _fixtures.FixtureTest, AssertsCompiledSQL):
             use_default_dialect=True,
         )
 
+    @testing.combinations(
+        # lambda
+    )
+    def test_join_syntaxes(self, expr):
+        User, Order, Item, Keyword = self._user_order_item_fixture()
+
     def test_transient_count(self):
         User, Address = self._user_address_fixture()
         u1 = User()
@@ -462,67 +521,6 @@ class DynamicTest(_DynamicFixture, _fixtures.FixtureTest, AssertsCompiledSQL):
         u1.addresses.append(Address())
         eq_(u1.addresses[0], Address())
 
-    def test_custom_query(self):
-        class MyQuery(Query):
-            pass
-
-        User, Address = self._user_address_fixture(
-            addresses_args={"query_class": MyQuery}
-        )
-
-        sess = create_session()
-        u = User()
-        sess.add(u)
-
-        col = u.addresses
-        assert isinstance(col, Query)
-        assert isinstance(col, MyQuery)
-        assert hasattr(col, "append")
-        eq_(type(col).__name__, "AppenderMyQuery")
-
-        q = col.limit(1)
-        assert isinstance(q, Query)
-        assert isinstance(q, MyQuery)
-        assert not hasattr(q, "append")
-        eq_(type(q).__name__, "MyQuery")
-
-    def test_custom_query_with_custom_mixin(self):
-        class MyAppenderMixin(AppenderMixin):
-            def add(self, items):
-                if isinstance(items, list):
-                    for item in items:
-                        self.append(item)
-                else:
-                    self.append(items)
-
-        class MyQuery(Query):
-            pass
-
-        class MyAppenderQuery(MyAppenderMixin, MyQuery):
-            query_class = MyQuery
-
-        User, Address = self._user_address_fixture(
-            addresses_args={"query_class": MyAppenderQuery}
-        )
-
-        sess = create_session()
-        u = User()
-        sess.add(u)
-
-        col = u.addresses
-        assert isinstance(col, Query)
-        assert isinstance(col, MyQuery)
-        assert hasattr(col, "append")
-        assert hasattr(col, "add")
-        eq_(type(col).__name__, "MyAppenderQuery")
-
-        q = col.limit(1)
-        assert isinstance(q, Query)
-        assert isinstance(q, MyQuery)
-        assert not hasattr(q, "append")
-        assert not hasattr(q, "add")
-        eq_(type(q).__name__, "MyQuery")
-
 
 class UOWTest(
     _DynamicFixture, _fixtures.FixtureTest, testing.AssertsExecutionResults
diff --git a/test/orm/test_froms.py b/test/orm/test_froms.py
index 1f09dc9e8..4b4c2bf73 100644
--- a/test/orm/test_froms.py
+++ b/test/orm/test_froms.py
@@ -1212,6 +1212,38 @@ class InstancesTest(QueryTest, AssertsCompiledSQL):
 
         self.assert_sql_count(testing.db, go, 1)
 
+    def test_contains_eager_four_future(self):
+        users, addresses, User = (
+            self.tables.users,
+            self.tables.addresses,
+            self.classes.User,
+        )
+
+        sess = create_session(future=True)
+
+        selectquery = users.outerjoin(addresses).select(
+            users.c.id < 10,
+            use_labels=True,
+            order_by=[users.c.id, addresses.c.id],
+        )
+
+        q = select(User)
+
+        def go():
+            result = (
+                sess.execute(
+                    q.options(contains_eager("addresses")).from_statement(
+                        selectquery
+                    )
+                )
+                .scalars()
+                .unique()
+                .all()
+            )
+            assert self.static.user_address_result[0:3] == result
+
+        self.assert_sql_count(testing.db, go, 1)
+
     def test_contains_eager_aliased(self):
         User, Address = self.classes.User, self.classes.Address
 
@@ -2122,14 +2154,17 @@ class MixedEntitiesTest(QueryTest, AssertsCompiledSQL):
             (user10, None),
         ]
 
-        sess = create_session()
+        sess = create_session(future=True)
 
         selectquery = users.outerjoin(addresses).select(
             use_labels=True, order_by=[users.c.id, addresses.c.id]
         )
+
+        result = sess.execute(
+            select(User, Address).from_statement(selectquery)
+        )
         eq_(
-            list(sess.query(User, Address).from_statement(selectquery)),
-            expected,
+            list(result), expected,
         )
         sess.expunge_all()
 
diff --git a/test/orm/test_options.py b/test/orm/test_options.py
index 74a8677bf..208db9d85 100644
--- a/test/orm/test_options.py
+++ b/test/orm/test_options.py
@@ -85,7 +85,7 @@ class PathTest(object):
                         ent.entity_zero
                         for ent in q._compile_state()._mapper_entities
                     ],
-                    q.compile_options._current_path,
+                    q._compile_options._current_path,
                     attr,
                     False,
                 )
@@ -1432,7 +1432,7 @@ class PickleTest(PathTest, QueryTest):
                 ent.entity_zero
                 for ent in query._compile_state()._mapper_entities
             ],
-            query.compile_options._current_path,
+            query._compile_options._current_path,
             attr,
             False,
         )
@@ -1469,7 +1469,7 @@ class PickleTest(PathTest, QueryTest):
                 ent.entity_zero
                 for ent in query._compile_state()._mapper_entities
             ],
-            query.compile_options._current_path,
+            query._compile_options._current_path,
             attr,
             False,
         )
@@ -1514,7 +1514,7 @@ class LocalOptsTest(PathTest, QueryTest):
                 for tb in opt._to_bind:
                     tb._bind_loader(
                         [ent.entity_zero for ent in ctx._mapper_entities],
-                        query.compile_options._current_path,
+                        query._compile_options._current_path,
                         attr,
                         False,
                     )
@@ -1608,7 +1608,7 @@ class SubOptionsTest(PathTest, QueryTest):
                     ent.entity_zero
                     for ent in q._compile_state()._mapper_entities
                 ],
-                q.compile_options._current_path,
+                q._compile_options._current_path,
                 attr_a,
                 False,
             )
@@ -1622,7 +1622,7 @@ class SubOptionsTest(PathTest, QueryTest):
                         ent.entity_zero
                         for ent in q._compile_state()._mapper_entities
                     ],
-                    q.compile_options._current_path,
+                    q._compile_options._current_path,
                     attr_b,
                     False,
                 )
diff --git a/test/orm/test_query.py b/test/orm/test_query.py
index 7ef2a455e..486254207 100644
--- a/test/orm/test_query.py
+++ b/test/orm/test_query.py
@@ -12,6 +12,7 @@ from sqlalchemy import collate
 from sqlalchemy import column
 from sqlalchemy import desc
 from sqlalchemy import distinct
+from sqlalchemy import event
 from sqlalchemy import exc as sa_exc
 from sqlalchemy import exists
 from sqlalchemy import ForeignKey
@@ -52,6 +53,7 @@ from sqlalchemy.orm import selectinload
 from sqlalchemy.orm import Session
 from sqlalchemy.orm import subqueryload
 from sqlalchemy.orm import synonym
+from sqlalchemy.orm.context import QueryContext
 from sqlalchemy.orm.util import join
 from sqlalchemy.orm.util import with_parent
 from sqlalchemy.sql import expression
@@ -560,6 +562,22 @@ class BindSensitiveStringifyTest(fixtures.TestBase):
 
 
 class GetTest(QueryTest):
+    def test_loader_options(self):
+        User = self.classes.User
+
+        s = Session()
+
+        u1 = s.query(User).options(joinedload(User.addresses)).get(8)
+        eq_(len(u1.__dict__["addresses"]), 3)
+
+    def test_loader_options_future(self):
+        User = self.classes.User
+
+        s = Session()
+
+        u1 = s.get(User, 8, options=[joinedload(User.addresses)])
+        eq_(len(u1.__dict__["addresses"]), 3)
+
     def test_get_composite_pk_keyword_based_no_result(self):
         CompositePk = self.classes.CompositePk
 
@@ -610,6 +628,18 @@ class GetTest(QueryTest):
         u2 = s.query(User).get(7)
         assert u is not u2
 
+    def test_get_future(self):
+        User = self.classes.User
+
+        s = create_session()
+        assert s.get(User, 19) is None
+        u = s.get(User, 7)
+        u2 = s.get(User, 7)
+        assert u is u2
+        s.expunge_all()
+        u2 = s.get(User, 7)
+        assert u is not u2
+
     def test_get_composite_pk_no_result(self):
         CompositePk = self.classes.CompositePk
 
@@ -843,6 +873,73 @@ class GetTest(QueryTest):
         assert u.addresses[0].email_address == "jack@bean.com"
         assert u.orders[1].items[2].description == "item 5"
 
+    def test_populate_existing_future(self):
+        User, Address = self.classes.User, self.classes.Address
+
+        s = Session(future=True, autoflush=False)
+
+        userlist = s.query(User).all()
+
+        u = userlist[0]
+        u.name = "foo"
+        a = Address(name="ed")
+        u.addresses.append(a)
+
+        self.assert_(a in u.addresses)
+
+        stmt = select(User).execution_options(populate_existing=True)
+
+        s.execute(stmt,).scalars().all()
+
+        self.assert_(u not in s.dirty)
+
+        self.assert_(u.name == "jack")
+
+        self.assert_(a not in u.addresses)
+
+        u.addresses[0].email_address = "lala"
+        u.orders[1].items[2].description = "item 12"
+        # test that lazy load doesn't change child items
+        s.query(User).populate_existing().all()
+        assert u.addresses[0].email_address == "lala"
+        assert u.orders[1].items[2].description == "item 12"
+
+        # eager load does
+
+        stmt = (
+            select(User)
+            .options(
+                joinedload("addresses"),
+                joinedload("orders").joinedload("items"),
+            )
+            .execution_options(populate_existing=True)
+        )
+
+        s.execute(stmt).scalars().all()
+
+        assert u.addresses[0].email_address == "jack@bean.com"
+        assert u.orders[1].items[2].description == "item 5"
+
+    def test_option_transfer_future(self):
+        User = self.classes.User
+        stmt = select(User).execution_options(
+            populate_existing=True, autoflush=False, yield_per=10
+        )
+        s = Session(testing.db, future=True)
+
+        m1 = mock.Mock()
+
+        event.listen(s, "do_orm_execute", m1)
+
+        s.execute(stmt)
+
+        eq_(
+            m1.mock_calls[0].args[0].load_options,
+            QueryContext.default_load_options(
+                _autoflush=False, _populate_existing=True, _yield_per=10
+            ),
+        )
+
 
 class InvalidGenerationsTest(QueryTest, AssertsCompiledSQL):
     @testing.combinations(
@@ -4339,6 +4436,31 @@ class TextTest(QueryTest, AssertsCompiledSQL):
             None,
         )
 
+    def test_select_star_future(self):
+        User = self.classes.User
+
+        sess = Session(future=True)
+        eq_(
+            sess.execute(
+                select(User).from_statement(
+                    text("select * from users order by id")
+                )
+            )
+            .scalars()
+            .first(),
+            User(id=7),
+        )
+        eq_(
+            sess.execute(
+                select(User).from_statement(
+                    text("select * from users where name='nonexistent'")
+                )
+            )
+            .scalars()
+            .first(),
+            None,
+        )
+
     def test_columns_mismatched(self):
         # test that columns using column._label match, as well as that
         # ordering doesn't matter
@@ -4360,6 +4482,27 @@ class TextTest(QueryTest, AssertsCompiledSQL):
             ],
         )
 
+    def test_columns_mismatched_future(self):
+        # test that columns using column._label match, as well as that
+        # ordering doesn't matter
+        User = self.classes.User
+
+        s = create_session(future=True)
+        q = select(User).from_statement(
+            text(
+                "select name, 27 as foo, id as users_id from users order by id"
+            )
+        )
+        eq_(
+            s.execute(q).scalars().all(),
+            [
+                User(id=7, name="jack"),
+                User(id=8, name="ed"),
+                User(id=9, name="fred"),
+                User(id=10, name="chuck"),
+            ],
+        )
+
     def test_columns_multi_table_uselabels(self):
         # test that columns using column._label match, as well as that
         # ordering doesn't matter.
@@ -4385,6 +4528,31 @@ class TextTest(QueryTest, AssertsCompiledSQL):
             ],
         )
 
+    def test_columns_multi_table_uselabels_future(self):
+        # test that columns using column._label match, as well as that
+        # ordering doesn't matter.
+        User = self.classes.User
+        Address = self.classes.Address
+
+        s = create_session(future=True)
+        q = select(User, Address).from_statement(
+            text(
+                "select users.name AS users_name, users.id AS users_id, "
+                "addresses.id AS addresses_id FROM users JOIN addresses "
+                "ON users.id = addresses.user_id WHERE users.id=8 "
+                "ORDER BY addresses.id"
+            )
+        )
+
+        eq_(
+            s.execute(q).all(),
+            [
+                (User(id=8), Address(id=2)),
+                (User(id=8), Address(id=3)),
+                (User(id=8), Address(id=4)),
+            ],
+        )
+
     def test_columns_multi_table_uselabels_contains_eager(self):
         # test that columns using column._label match, as well as that
         # ordering doesn't matter.
@@ -4411,6 +4579,32 @@ class TextTest(QueryTest, AssertsCompiledSQL):
 
         self.assert_sql_count(testing.db, go, 1)
 
+    def test_columns_multi_table_uselabels_contains_eager_future(self):
+        # test that columns using column._label match, as well as that
+        # ordering doesn't matter.
+        User = self.classes.User
+        Address = self.classes.Address
+
+        s = create_session(future=True)
+        q = (
+            select(User)
+            .from_statement(
+                text(
+                    "select users.name AS users_name, users.id AS users_id, "
+                    "addresses.id AS addresses_id FROM users JOIN addresses "
+                    "ON users.id = addresses.user_id WHERE users.id=8 "
+                    "ORDER BY addresses.id"
+                )
+            )
+            .options(contains_eager(User.addresses))
+        )
+
+        def go():
+            r = s.execute(q).unique().scalars().all()
+            eq_(r[0].addresses, [Address(id=2), Address(id=3), Address(id=4)])
+
+        self.assert_sql_count(testing.db, go, 1)
+
     def test_columns_multi_table_uselabels_cols_contains_eager(self):
         # test that columns using column._label match, as well as that
         # ordering doesn't matter.
@@ -4437,6 +4631,32 @@ class TextTest(QueryTest, AssertsCompiledSQL):
 
         self.assert_sql_count(testing.db, go, 1)
 
+    def test_columns_multi_table_uselabels_cols_contains_eager_future(self):
+        # test that columns using column._label match, as well as that
+        # ordering doesn't matter.
+        User = self.classes.User
+        Address = self.classes.Address
+
+        s = create_session(future=True)
+        q = (
+            select(User)
+            .from_statement(
+                text(
+                    "select users.name AS users_name, users.id AS users_id, "
+                    "addresses.id AS addresses_id FROM users JOIN addresses "
+                    "ON users.id = addresses.user_id WHERE users.id=8 "
+                    "ORDER BY addresses.id"
+                ).columns(User.name, User.id, Address.id)
+            )
+            .options(contains_eager(User.addresses))
+        )
+
+        def go():
+            r = s.execute(q).unique().scalars().all()
+            eq_(r[0].addresses, [Address(id=2), Address(id=3), Address(id=4)])
+
+        self.assert_sql_count(testing.db, go, 1)
+
     def test_textual_select_orm_columns(self):
         # test that columns using column._label match, as well as that
         # ordering doesn't matter.
@@ -4521,6 +4741,34 @@ class TextTest(QueryTest, AssertsCompiledSQL):
             [User(id=9)],
         )
 
+    def test_whereclause_future(self):
+        User = self.classes.User
+
+        s = create_session(future=True)
+        eq_(
+            s.execute(select(User).filter(text("id in (8, 9)")))
+            .scalars()
+            .all(),
+            [User(id=8), User(id=9)],
+        )
+
+        eq_(
+            s.execute(
+                select(User).filter(text("name='fred'")).filter(text("id=9"))
+            )
+            .scalars()
+            .all(),
+            [User(id=9)],
+        )
+        eq_(
+            s.execute(
+                select(User).filter(text("name='fred'")).filter(User.id == 9)
+            )
+            .scalars()
+            .all(),
+            [User(id=9)],
+        )
+
     def test_binds_coerce(self):
         User = self.classes.User
 
diff --git a/test/orm/test_relationships.py b/test/orm/test_relationships.py
index 702e1ea92..22b028c47 100644
--- a/test/orm/test_relationships.py
+++ b/test/orm/test_relationships.py
@@ -1345,19 +1345,13 @@ class CompositeSelfRefFKTest(fixtures.MappedTest, AssertsCompiledSQL):
         e1 = sess.query(Employee).filter_by(name="emp1").one()
         e5 = sess.query(Employee).filter_by(name="emp5").one()
 
-        test_e1 = sess.query(Employee).get([c1.company_id, e1.emp_id])
+        test_e1 = sess.get(Employee, [c1.company_id, e1.emp_id])
         assert test_e1.name == "emp1", test_e1.name
-        test_e5 = sess.query(Employee).get([c2.company_id, e5.emp_id])
+        test_e5 = sess.get(Employee, [c2.company_id, e5.emp_id])
         assert test_e5.name == "emp5", test_e5.name
         assert [x.name for x in test_e1.employees] == ["emp2", "emp3"]
-        assert (
-            sess.query(Employee).get([c1.company_id, 3]).reports_to.name
-            == "emp1"
-        )
-        assert (
-            sess.query(Employee).get([c2.company_id, 3]).reports_to.name
-            == "emp5"
-        )
+        assert sess.get(Employee, [c1.company_id, 3]).reports_to.name == "emp1"
+        assert sess.get(Employee, [c2.company_id, 3]).reports_to.name == "emp5"
 
     def _test_join_aliasing(self, sess):
         Employee = self.classes.Employee
diff --git a/test/orm/test_session.py b/test/orm/test_session.py
index d3d7990c5..143b577cf 100644
--- a/test/orm/test_session.py
+++ b/test/orm/test_session.py
@@ -174,8 +174,7 @@ class TransScopingTest(_fixtures.FixtureTest):
 
         assert_raises_message(
             sa.exc.InvalidRequestError,
-            "A transaction is already begun.  Use "
-            "subtransactions=True to allow subtransactions.",
+            "A transaction is already begun on this Session.",
             s.begin,
         )
 
@@ -625,7 +624,7 @@ class SessionStateTest(_fixtures.FixtureTest):
         session.flush()
         session.commit()
 
-    def test_active_flag(self):
+    def test_active_flag_autocommit(self):
         sess = create_session(bind=config.db, autocommit=True)
         assert not sess.is_active
         sess.begin()
@@ -633,6 +632,37 @@ class SessionStateTest(_fixtures.FixtureTest):
         sess.rollback()
         assert not sess.is_active
 
+    def test_active_flag_autobegin(self):
+        sess = create_session(bind=config.db, autocommit=False)
+        assert sess.is_active
+        assert not sess.in_transaction()
+        sess.begin()
+        assert sess.is_active
+        sess.rollback()
+        assert sess.is_active
+
+    def test_active_flag_autobegin_future(self):
+        sess = create_session(bind=config.db, future=True)
+        assert sess.is_active
+        assert not sess.in_transaction()
+        sess.begin()
+        assert sess.is_active
+        sess.rollback()
+        assert sess.is_active
+
+    def test_active_flag_partial_rollback(self):
+        sess = create_session(bind=config.db, autocommit=False)
+        assert sess.is_active
+        assert not sess.in_transaction()
+        sess.begin()
+        assert sess.is_active
+        sess.begin(_subtrans=True)
+        sess.rollback()
+        assert not sess.is_active
+
+        sess.rollback()
+        assert sess.is_active
+
     @engines.close_open_connections
     def test_add_delete(self):
         User, Address, addresses, users = (
diff --git a/test/orm/test_transaction.py b/test/orm/test_transaction.py
index 92bc634b5..660bc7a5d 100644
--- a/test/orm/test_transaction.py
+++ b/test/orm/test_transaction.py
@@ -1,8 +1,12 @@
+import contextlib
+
 from sqlalchemy import Column
 from sqlalchemy import event
 from sqlalchemy import exc as sa_exc
 from sqlalchemy import func
 from sqlalchemy import inspect
+from sqlalchemy import Integer
+from sqlalchemy import MetaData
 from sqlalchemy import select
 from sqlalchemy import String
 from sqlalchemy import Table
@@ -27,7 +31,6 @@ from sqlalchemy.testing import expect_warnings
 from sqlalchemy.testing import fixtures
 from sqlalchemy.testing import is_
 from sqlalchemy.testing import is_not_
-from sqlalchemy.testing import is_true
 from sqlalchemy.testing import mock
 from sqlalchemy.testing.util import gc_collect
 from test.orm._fixtures import FixtureTest
@@ -107,11 +110,11 @@ class SessionTransactionTest(fixtures.RemovesEvents, FixtureTest):
             sess.add(u1)
             sess.flush()
 
-            sess.begin_nested()
+            savepoint = sess.begin_nested()
             u2 = User(name="u2")
             sess.add(u2)
             sess.flush()
-            sess.rollback()
+            savepoint.rollback()
 
             trans.commit()
             assert len(sess.query(User).all()) == 1
@@ -259,7 +262,7 @@ class SessionTransactionTest(fixtures.RemovesEvents, FixtureTest):
         users = self.tables.users
 
         engine = Engine._future_facade(testing.db)
-        session = create_session(engine)
+        session = create_session(engine, autocommit=False)
 
         session.begin()
         session.connection().execute(users.insert().values(name="user1"))
@@ -316,6 +319,34 @@ class SessionTransactionTest(fixtures.RemovesEvents, FixtureTest):
         assert attributes.instance_state(u1).expired
         eq_(u1.name, "u1")
 
+    @testing.requires.savepoints
+    def test_dirty_state_transferred_deep_nesting_future(self):
+        User, users = self.classes.User, self.tables.users
+
+        mapper(User, users)
+
+        s = Session(testing.db, future=True)
+        u1 = User(name="u1")
+        s.add(u1)
+        s.commit()
+
+        nt1 = s.begin_nested()
+        nt2 = s.begin_nested()
+        u1.name = "u2"
+        assert attributes.instance_state(u1) not in nt2._dirty
+        assert attributes.instance_state(u1) not in nt1._dirty
+        s.flush()
+        assert attributes.instance_state(u1) in nt2._dirty
+        assert attributes.instance_state(u1) not in nt1._dirty
+
+        nt2.commit()
+        assert attributes.instance_state(u1) in nt2._dirty
+        assert attributes.instance_state(u1) in nt1._dirty
+
+        nt1.rollback()
+        assert attributes.instance_state(u1).expired
+        eq_(u1.name, "u1")
+
     @testing.requires.independent_connections
     def test_transactions_isolated(self):
         User, users = self.classes.User, self.tables.users
@@ -441,13 +472,35 @@ class SessionTransactionTest(fixtures.RemovesEvents, FixtureTest):
         sess.add(u2)
         sess.flush()
 
-        sess.rollback()
+        sess.rollback()  # rolls back nested only
 
         sess.commit()
         assert len(sess.query(User).all()) == 1
         sess.close()
 
     @testing.requires.savepoints
+    def test_nested_autotrans_future(self):
+        User, users = self.classes.User, self.tables.users
+
+        mapper(User, users)
+        sess = create_session(autocommit=False, future=True)
+        u = User(name="u1")
+        sess.add(u)
+        sess.flush()
+
+        sess.begin_nested()  # nested transaction
+
+        u2 = User(name="u2")
+        sess.add(u2)
+        sess.flush()
+
+        sess.rollback()  # rolls back the whole trans
+
+        sess.commit()
+        assert len(sess.query(User).all()) == 0
+        sess.close()
+
+    @testing.requires.savepoints
     def test_nested_transaction_connection_add(self):
         users, User = self.tables.users, self.classes.User
 
@@ -726,7 +779,7 @@ class SessionTransactionTest(fixtures.RemovesEvents, FixtureTest):
             testing.db.dialect, "do_execute", prevent_savepoint_rollback
         )
 
-        with session.transaction:
+        with session.begin():
             session.add(User(id=1, name="x"))
 
         session.begin_nested()
@@ -739,13 +792,15 @@ class SessionTransactionTest(fixtures.RemovesEvents, FixtureTest):
         )
 
         # rollback succeeds, because the Session is deactivated
-        eq_(session.transaction._state, _session.DEACTIVE)
+        eq_(session._transaction._state, _session.DEACTIVE)
+        eq_(session.is_active, False)
         session.rollback()
 
         # back to normal
-        eq_(session.transaction._state, _session.ACTIVE)
+        eq_(session._transaction._state, _session.ACTIVE)
+        eq_(session.is_active, True)
 
-        trans = session.transaction
+        trans = session._transaction
 
         # leave the outermost trans
         session.rollback()
@@ -754,75 +809,10 @@ class SessionTransactionTest(fixtures.RemovesEvents, FixtureTest):
         eq_(trans._state, _session.CLOSED)
 
         # outermost transaction is new
-        is_not_(session.transaction, trans)
-
-        # outermost is active
-        eq_(session.transaction._state, _session.ACTIVE)
-
-    @testing.requires.independent_connections
-    @testing.emits_warning(".*previous exception")
-    def test_failed_rollback_deactivates_transaction_ctx_integration(self):
-        # test #4050 in the same context as that of oslo.db
-
-        users, User = self.tables.users, self.classes.User
-
-        mapper(User, users)
-        session = Session(bind=testing.db, autocommit=True)
-
-        evented_exceptions = []
-        caught_exceptions = []
-
-        def canary(context):
-            evented_exceptions.append(context.original_exception)
-
-        rollback_error = testing.db.dialect.dbapi.InterfaceError(
-            "Can't roll back to savepoint"
-        )
-
-        def prevent_savepoint_rollback(
-            cursor, statement, parameters, context=None
-        ):
-            if (
-                context is not None
-                and context.compiled
-                and isinstance(
-                    context.compiled.statement,
-                    elements.RollbackToSavepointClause,
-                )
-            ):
-                raise rollback_error
+        is_not_(session._transaction, trans)
 
-        self.event_listen(testing.db, "handle_error", canary, retval=True)
-        self.event_listen(
-            testing.db.dialect, "do_execute", prevent_savepoint_rollback
-        )
-
-        with session.begin():
-            session.add(User(id=1, name="x"))
-
-        try:
-            with session.begin():
-                try:
-                    with session.begin_nested():
-                        # raises IntegrityError on flush
-                        session.add(User(id=1, name="x"))
-
-                # outermost is the failed SAVEPOINT rollback
-                # from the "with session.begin_nested()"
-                except sa_exc.DBAPIError as dbe_inner:
-                    caught_exceptions.append(dbe_inner.orig)
-                    raise
-        except sa_exc.DBAPIError as dbe_outer:
-            caught_exceptions.append(dbe_outer.orig)
-
-        is_true(
-            isinstance(
-                evented_exceptions[0], testing.db.dialect.dbapi.IntegrityError
-            )
-        )
-        eq_(evented_exceptions[1], rollback_error)
-        eq_(len(evented_exceptions), 2)
-        eq_(caught_exceptions, [rollback_error, rollback_error])
+        is_(session._transaction, None)
+        eq_(session.is_active, True)
 
     def test_no_prepare_wo_twophase(self):
         sess = create_session(bind=testing.db, autocommit=False)
@@ -905,6 +895,7 @@ class SessionTransactionTest(fixtures.RemovesEvents, FixtureTest):
                 )
             )
         )
+
         sess = Session(bind=bind)
         c1 = sess.connection(execution_options={"isolation_level": "FOO"})
         eq_(bind.mock_calls, [mock.call.connect()])
@@ -912,7 +903,7 @@ class SessionTransactionTest(fixtures.RemovesEvents, FixtureTest):
             bind.connect().mock_calls,
             [mock.call.execution_options(isolation_level="FOO")],
         )
-        eq_(bind.connect().execution_options().mock_calls, [mock.call.begin()])
+
         eq_(c1, bind.connect().execution_options())
 
     def test_execution_options_ignored_mid_transaction(self):
@@ -1034,9 +1025,19 @@ class SessionTransactionTest(fixtures.RemovesEvents, FixtureTest):
         session = create_session(autocommit=False)
         session.add(User(name="ed"))
         session.transaction.commit()
-        assert (
-            session.transaction is not None
-        ), "autocommit=False should start a new transaction"
+
+        is_not_(session.transaction, None)
+
+    def test_no_autocommit_with_explicit_commit_future(self):
+        User, users = self.classes.User, self.tables.users
+
+        mapper(User, users)
+        session = create_session(autocommit=False, future=True)
+        session.add(User(name="ed"))
+        session.transaction.commit()
+
+        # new in 1.4
+        is_(session.transaction, None)
 
     @testing.requires.python2
     @testing.requires.savepoints_w_release
@@ -1088,6 +1089,205 @@ class _LocalFixture(FixtureTest):
         mapper(Address, addresses)
 
 
+class SubtransactionRecipeTest(FixtureTest):
+    run_inserts = None
+    __backend__ = True
+
+    future = False
+
+    @testing.fixture
+    def subtransaction_recipe(self):
+        @contextlib.contextmanager
+        def transaction(session):
+
+            if session.in_transaction():
+                outermost = False
+            else:
+                outermost = True
+                session.begin()
+
+            try:
+                yield
+            except:
+                if session.in_transaction():
+                    session.rollback()
+                raise
+            else:
+                if outermost and session.in_transaction():
+                    session.commit()
+
+        return transaction
+
+    @testing.requires.savepoints
+    def test_recipe_heavy_nesting(self, subtransaction_recipe):
+        users = self.tables.users
+
+        session = Session(testing.db, future=self.future)
+
+        with subtransaction_recipe(session):
+            session.connection().execute(users.insert().values(name="user1"))
+            with subtransaction_recipe(session):
+                savepoint = session.begin_nested()
+                session.connection().execute(
+                    users.insert().values(name="user2")
+                )
+                assert (
+                    session.connection()
+                    .exec_driver_sql("select count(1) from users")
+                    .scalar()
+                    == 2
+                )
+                savepoint.rollback()
+
+            with subtransaction_recipe(session):
+                assert (
+                    session.connection()
+                    .exec_driver_sql("select count(1) from users")
+                    .scalar()
+                    == 1
+                )
+                session.connection().execute(
+                    users.insert().values(name="user3")
+                )
+            assert (
+                session.connection()
+                .exec_driver_sql("select count(1) from users")
+                .scalar()
+                == 2
+            )
+
+    @engines.close_open_connections
+    def test_recipe_subtransaction_on_external_subtrans(
+        self, subtransaction_recipe
+    ):
+        users, User = self.tables.users, self.classes.User
+
+        mapper(User, users)
+        conn = testing.db.connect()
+        trans = conn.begin()
+        sess = Session(conn, future=self.future)
+
+        with subtransaction_recipe(sess):
+            u = User(name="ed")
+            sess.add(u)
+            sess.flush()
+            # commit does nothing
+        trans.rollback()  # rolls back
+        assert len(sess.query(User).all()) == 0
+        sess.close()
+
+    def test_recipe_commit_one(self, subtransaction_recipe):
+        User, users = self.classes.User, self.tables.users
+
+        mapper(User, users)
+        sess = Session(testing.db, future=self.future)
+
+        with subtransaction_recipe(sess):
+            u = User(name="u1")
+            sess.add(u)
+        sess.close()
+        assert len(sess.query(User).all()) == 1
+
+    def test_recipe_subtransaction_on_noautocommit(
+        self, subtransaction_recipe
+    ):
+        User, users = self.classes.User, self.tables.users
+
+        mapper(User, users)
+        sess = Session(testing.db, future=self.future)
+
+        sess.begin()
+        with subtransaction_recipe(sess):
+            u = User(name="u1")
+            sess.add(u)
+            sess.flush()
+        sess.rollback()  # rolls back
+        assert len(sess.query(User).all()) == 0
+        sess.close()
+
+    @testing.requires.savepoints
+    def test_recipe_mixed_transaction_control(self, subtransaction_recipe):
+        users, User = self.tables.users, self.classes.User
+
+        mapper(User, users)
+
+        sess = Session(testing.db, future=self.future)
+
+        sess.begin()
+        sess.begin_nested()
+
+        with subtransaction_recipe(sess):
+
+            sess.add(User(name="u1"))
+
+        sess.commit()
+        sess.commit()
+
+        eq_(len(sess.query(User).all()), 1)
+        sess.close()
+
+        t1 = sess.begin()
+        t2 = sess.begin_nested()
+
+        sess.add(User(name="u2"))
+
+        t2.commit()
+        assert sess.transaction is t1
+
+        sess.close()
+
+    def test_recipe_error_on_using_inactive_session_commands(
+        self, subtransaction_recipe
+    ):
+        users, User = self.tables.users, self.classes.User
+
+        mapper(User, users)
+        sess = Session(testing.db, future=self.future)
+        sess.begin()
+
+        try:
+            with subtransaction_recipe(sess):
+                sess.add(User(name="u1"))
+                sess.flush()
+                raise Exception("force rollback")
+        except:
+            pass
+
+        # that was a real rollback, so no transaction
+        is_(sess.get_transaction(), None)
+
+        sess.close()
+
+    def test_recipe_multi_nesting(self, subtransaction_recipe):
+        sess = Session(testing.db, future=self.future)
+
+        with subtransaction_recipe(sess):
+            assert sess.in_transaction()
+
+            try:
+                with subtransaction_recipe(sess):
+                    assert sess.transaction
+                    raise Exception("force rollback")
+            except:
+                pass
+
+            assert not sess.in_transaction()
+
+    def test_recipe_deactive_status_check(self, subtransaction_recipe):
+        sess = Session(testing.db, future=self.future)
+        sess.begin()
+
+        with subtransaction_recipe(sess):
+            sess.rollback()
+
+        assert not sess.in_transaction()
+        sess.commit()  # no error
+
+
+class FutureSubtransactionRecipeTest(SubtransactionRecipeTest):
+    future = True
+
+
 class FixtureDataTest(_LocalFixture):
     run_inserts = "each"
     __backend__ = True
@@ -1135,23 +1335,31 @@ class CleanSavepointTest(FixtureTest):
     run_inserts = None
     __backend__ = True
 
-    def _run_test(self, update_fn):
+    def _run_test(self, update_fn, future=False):
         User, users = self.classes.User, self.tables.users
 
         mapper(User, users)
 
-        s = Session(bind=testing.db)
+        s = Session(bind=testing.db, future=future)
         u1 = User(name="u1")
         u2 = User(name="u2")
         s.add_all([u1, u2])
         s.commit()
         u1.name
         u2.name
+        trans = s._transaction
+        assert trans is not None
         s.begin_nested()
         update_fn(s, u2)
         eq_(u2.name, "u2modified")
         s.rollback()
-        eq_(u1.__dict__["name"], "u1")
+
+        if future:
+            assert s._transaction is None
+            assert "name" not in u1.__dict__
+        else:
+            assert s._transaction is trans
+            eq_(u1.__dict__["name"], "u1")
         assert "name" not in u2.__dict__
         eq_(u2.name, "u2")
 
@@ -1185,64 +1393,6 @@ class CleanSavepointTest(FixtureTest):
         self._run_test(update_fn)
 
 
-class ContextManagerTest(FixtureTest):
-    run_inserts = None
-    __backend__ = True
-
-    @testing.requires.savepoints
-    @engines.close_open_connections
-    def test_contextmanager_nested_rollback(self):
-        users, User = self.tables.users, self.classes.User
-
-        mapper(User, users)
-
-        sess = Session()
-
-        def go():
-            with sess.begin_nested():
-                sess.add(User())  # name can't be null
-                sess.flush()
-
-        # and not InvalidRequestError
-        assert_raises(sa_exc.DBAPIError, go)
-
-        with sess.begin_nested():
-            sess.add(User(name="u1"))
-
-        eq_(sess.query(User).count(), 1)
-
-    def test_contextmanager_commit(self):
-        users, User = self.tables.users, self.classes.User
-
-        mapper(User, users)
-
-        sess = Session(autocommit=True)
-        with sess.begin():
-            sess.add(User(name="u1"))
-
-        sess.rollback()
-        eq_(sess.query(User).count(), 1)
-
-    def test_contextmanager_rollback(self):
-        users, User = self.tables.users, self.classes.User
-
-        mapper(User, users)
-
-        sess = Session(autocommit=True)
-
-        def go():
-            with sess.begin():
-                sess.add(User())  # name can't be null
-
-        assert_raises(sa_exc.DBAPIError, go)
-
-        eq_(sess.query(User).count(), 0)
-
-        with sess.begin():
-            sess.add(User(name="u1"))
-        eq_(sess.query(User).count(), 1)
-
-
 class AutoExpireTest(_LocalFixture):
     __backend__ = True
 
@@ -1855,6 +2005,442 @@ class AutoCommitTest(_LocalFixture):
         eq_(u1.id, 3)
 
 
+class ContextManagerPlusFutureTest(FixtureTest):
+    run_inserts = None
+    __backend__ = True
+
+    @testing.requires.savepoints
+    @engines.close_open_connections
+    def test_contextmanager_nested_rollback(self):
+        users, User = self.tables.users, self.classes.User
+
+        mapper(User, users)
+
+        sess = Session()
+
+        def go():
+            with sess.begin_nested():
+                sess.add(User())  # name can't be null
+                sess.flush()
+
+        # and not InvalidRequestError
+        assert_raises(sa_exc.DBAPIError, go)
+
+        with sess.begin_nested():
+            sess.add(User(name="u1"))
+
+        eq_(sess.query(User).count(), 1)
+
+    def test_contextmanager_commit(self):
+        users, User = self.tables.users, self.classes.User
+
+        mapper(User, users)
+
+        sess = Session()
+        with sess.begin():
+            sess.add(User(name="u1"))
+
+        sess.rollback()
+        eq_(sess.query(User).count(), 1)
+
+    def test_contextmanager_rollback(self):
+        users, User = self.tables.users, self.classes.User
+
+        mapper(User, users)
+
+        sess = Session()
+
+        def go():
+            with sess.begin():
+                sess.add(User())  # name can't be null
+
+        assert_raises(sa_exc.DBAPIError, go)
+
+        eq_(sess.query(User).count(), 0)
+        sess.close()
+
+        with sess.begin():
+            sess.add(User(name="u1"))
+        eq_(sess.query(User).count(), 1)
+
+    def test_explicit_begin(self):
+        s1 = Session(testing.db)
+        with s1.begin() as trans:
+            is_(trans, s1.transaction)
+            s1.connection()
+
+        is_(s1._transaction, None)
+
+    def test_no_double_begin_explicit(self):
+        s1 = Session(testing.db)
+        s1.begin()
+        assert_raises_message(
+            sa_exc.InvalidRequestError,
+            "A transaction is already begun on this Session.",
+            s1.begin,
+        )
+
+    @testing.requires.savepoints
+    def test_future_rollback_is_global(self):
+        users = self.tables.users
+
+        s1 = Session(testing.db, future=True)
+
+        s1.begin()
+
+        s1.connection().execute(users.insert(), [{"id": 1, "name": "n1"}])
+
+        s1.begin_nested()
+
+        s1.connection().execute(
+            users.insert(), [{"id": 2, "name": "n2"}, {"id": 3, "name": "n3"}]
+        )
+
+        eq_(s1.connection().scalar(select(func.count()).select_from(users)), 3)
+
+        # rolls back the whole transaction
+        s1.rollback()
+        is_(s1.transaction, None)
+
+        eq_(s1.connection().scalar(select(func.count()).select_from(users)), 0)
+
+        s1.commit()
+        is_(s1.transaction, None)
+
+    @testing.requires.savepoints
+    def test_old_rollback_is_local(self):
+        users = self.tables.users
+
+        s1 = Session(testing.db)
+
+        t1 = s1.begin()
+
+        s1.connection().execute(users.insert(), [{"id": 1, "name": "n1"}])
+
+        s1.begin_nested()
+
+        s1.connection().execute(
+            users.insert(), [{"id": 2, "name": "n2"}, {"id": 3, "name": "n3"}]
+        )
+
+        eq_(s1.connection().scalar(select(func.count()).select_from(users)), 3)
+
+        # rolls back only the savepoint
+        s1.rollback()
+
+        is_(s1.transaction, t1)
+
+        eq_(s1.connection().scalar(select(func.count()).select_from(users)), 1)
+
+        s1.commit()
+        eq_(s1.connection().scalar(select(func.count()).select_from(users)), 1)
+        is_not_(s1.transaction, None)
+
+    def test_session_as_ctx_manager_one(self):
+        users = self.tables.users
+
+        with Session(testing.db) as sess:
+            is_not_(sess.transaction, None)
+
+            sess.connection().execute(
+                users.insert().values(id=1, name="user1")
+            )
+
+            eq_(
+                sess.connection().execute(users.select()).all(), [(1, "user1")]
+            )
+
+            is_not_(sess.transaction, None)
+
+        is_not_(sess.transaction, None)
+
+        # did not commit
+        eq_(sess.connection().execute(users.select()).all(), [])
+
+    def test_session_as_ctx_manager_future_one(self):
+        users = self.tables.users
+
+        with Session(testing.db, future=True) as sess:
+            is_(sess.transaction, None)
+
+            sess.connection().execute(
+                users.insert().values(id=1, name="user1")
+            )
+
+            eq_(
+                sess.connection().execute(users.select()).all(), [(1, "user1")]
+            )
+
+            is_not_(sess.transaction, None)
+
+        is_(sess.transaction, None)
+
+        # did not commit
+        eq_(sess.connection().execute(users.select()).all(), [])
+
+    def test_session_as_ctx_manager_two(self):
+        users = self.tables.users
+
+        try:
+            with Session(testing.db) as sess:
+                is_not_(sess.transaction, None)
+
+                sess.connection().execute(
+                    users.insert().values(id=1, name="user1")
+                )
+
+                raise Exception("force rollback")
+        except:
+            pass
+        is_not_(sess.transaction, None)
+
+    def test_session_as_ctx_manager_two_future(self):
+        users = self.tables.users
+
+        try:
+            with Session(testing.db, future=True) as sess:
+                is_(sess.transaction, None)
+
+                sess.connection().execute(
+                    users.insert().values(id=1, name="user1")
+                )
+
+                raise Exception("force rollback")
+        except:
+            pass
+        is_(sess.transaction, None)
+
+    def test_begin_context_manager(self):
+        users = self.tables.users
+
+        with Session(testing.db) as sess:
+            with sess.begin():
+                sess.connection().execute(
+                    users.insert().values(id=1, name="user1")
+                )
+
+                eq_(
+                    sess.connection().execute(users.select()).all(),
+                    [(1, "user1")],
+                )
+
+        # committed
+        eq_(sess.connection().execute(users.select()).all(), [(1, "user1")])
+
+    def test_sessionmaker_begin_context_manager(self):
+        users = self.tables.users
+
+        session = sessionmaker(testing.db)
+
+        with session.begin() as sess:
+            sess.connection().execute(
+                users.insert().values(id=1, name="user1")
+            )
+
+            eq_(
+                sess.connection().execute(users.select()).all(),
+                [(1, "user1")],
+            )
+
+        # committed
+        eq_(sess.connection().execute(users.select()).all(), [(1, "user1")])
+
+    def test_begin_context_manager_rollback_trans(self):
+        users = self.tables.users
+
+        try:
+            with Session(testing.db) as sess:
+                with sess.begin():
+                    sess.connection().execute(
+                        users.insert().values(id=1, name="user1")
+                    )
+
+                    eq_(
+                        sess.connection().execute(users.select()).all(),
+                        [(1, "user1")],
+                    )
+
+                    raise Exception("force rollback")
+        except:
+            pass
+
+        # rolled back
+        eq_(sess.connection().execute(users.select()).all(), [])
+
+    def test_begin_context_manager_rollback_outer(self):
+        users = self.tables.users
+
+        try:
+            with Session(testing.db) as sess:
+                with sess.begin():
+                    sess.connection().execute(
+                        users.insert().values(id=1, name="user1")
+                    )
+
+                    eq_(
+                        sess.connection().execute(users.select()).all(),
+                        [(1, "user1")],
+                    )
+
+                raise Exception("force rollback")
+        except:
+            pass
+
+        # committed
+        eq_(sess.connection().execute(users.select()).all(), [(1, "user1")])
+
+    def test_sessionmaker_begin_context_manager_rollback_trans(self):
+        users = self.tables.users
+
+        session = sessionmaker(testing.db)
+
+        try:
+            with session.begin() as sess:
+                sess.connection().execute(
+                    users.insert().values(id=1, name="user1")
+                )
+
+                eq_(
+                    sess.connection().execute(users.select()).all(),
+                    [(1, "user1")],
+                )
+
+                raise Exception("force rollback")
+        except:
+            pass
+
+        # rolled back
+        eq_(sess.connection().execute(users.select()).all(), [])
+
+    def test_sessionmaker_begin_context_manager_rollback_outer(self):
+        users = self.tables.users
+
+        session = sessionmaker(testing.db)
+
+        try:
+            with session.begin() as sess:
+                sess.connection().execute(
+                    users.insert().values(id=1, name="user1")
+                )
+
+                eq_(
+                    sess.connection().execute(users.select()).all(),
+                    [(1, "user1")],
+                )
+
+            raise Exception("force rollback")
+        except:
+            pass
+
+        # committed
+        eq_(sess.connection().execute(users.select()).all(), [(1, "user1")])
+
+
+class TransactionFlagsTest(fixtures.TestBase):
+    def test_in_transaction(self):
+        s1 = Session(testing.db)
+
+        eq_(s1.in_transaction(), False)
+
+        trans = s1.begin()
+
+        eq_(s1.in_transaction(), True)
+        is_(s1.get_transaction(), trans)
+
+        n1 = s1.begin_nested()
+
+        eq_(s1.in_transaction(), True)
+        is_(s1.get_transaction(), trans)
+        is_(s1.get_nested_transaction(), n1)
+
+        n1.rollback()
+
+        is_(s1.get_nested_transaction(), None)
+        is_(s1.get_transaction(), trans)
+
+        eq_(s1.in_transaction(), True)
+
+        s1.commit()
+
+        eq_(s1.in_transaction(), False)
+        is_(s1.get_transaction(), None)
+
+    def test_in_transaction_subtransactions(self):
+        """we'd like to do away with subtransactions for future sessions
+        entirely.  at the moment we are still using them internally.
+        it might be difficult to keep the internals working in exactly
+        the same way if remove this concept, so for now just test that
+        the external API works.
+
+        """
+        s1 = Session(testing.db)
+
+        eq_(s1.in_transaction(), False)
+
+        trans = s1.begin()
+
+        eq_(s1.in_transaction(), True)
+        is_(s1.get_transaction(), trans)
+
+        subtrans = s1.begin(_subtrans=True)
+        is_(s1.get_transaction(), trans)
+        eq_(s1.in_transaction(), True)
+
+        is_(s1._transaction, subtrans)
+
+        s1.rollback()
+
+        eq_(s1.in_transaction(), True)
+        is_(s1._transaction, trans)
+
+        s1.rollback()
+
+        eq_(s1.in_transaction(), False)
+        is_(s1._transaction, None)
+
+    def test_in_transaction_nesting(self):
+        s1 = Session(testing.db)
+
+        eq_(s1.in_transaction(), False)
+
+        trans = s1.begin()
+
+        eq_(s1.in_transaction(), True)
+        is_(s1.get_transaction(), trans)
+
+        sp1 = s1.begin_nested()
+
+        eq_(s1.in_transaction(), True)
+        is_(s1.get_transaction(), trans)
+        is_(s1.get_nested_transaction(), sp1)
+
+        sp2 = s1.begin_nested()
+
+        eq_(s1.in_transaction(), True)
+        eq_(s1.in_nested_transaction(), True)
+        is_(s1.get_transaction(), trans)
+        is_(s1.get_nested_transaction(), sp2)
+
+        sp2.rollback()
+
+        eq_(s1.in_transaction(), True)
+        eq_(s1.in_nested_transaction(), True)
+        is_(s1.get_transaction(), trans)
+        is_(s1.get_nested_transaction(), sp1)
+
+        sp1.rollback()
+
+        is_(s1.get_nested_transaction(), None)
+        eq_(s1.in_transaction(), True)
+        eq_(s1.in_nested_transaction(), False)
+        is_(s1.get_transaction(), trans)
+
+        s1.rollback()
+
+        eq_(s1.in_transaction(), False)
+        is_(s1.get_transaction(), None)
+
+
 class NaturalPKRollbackTest(fixtures.MappedTest):
     __backend__ = True
 
@@ -2031,3 +2617,199 @@ class NaturalPKRollbackTest(fixtures.MappedTest):
         assert u2 not in s
 
         assert s.identity_map[identity_key(User, ("u1",))] is u1
+
+
+class JoinIntoAnExternalTransactionFixture(object):
+    """Test the "join into an external transaction" examples"""
+
+    def setup(self):
+        self.connection = testing.db.connect()
+
+        self.metadata = MetaData()
+        self.table = Table(
+            "t1", self.metadata, Column("id", Integer, primary_key=True)
+        )
+        with self.connection.begin():
+            self.table.create(self.connection, checkfirst=True)
+
+        self.setup_session()
+
+    def test_something(self):
+        A = self.A
+
+        a1 = A()
+        self.session.add(a1)
+        self.session.commit()
+
+        self._assert_count(1)
+
+    @testing.requires.savepoints
+    def test_something_with_rollback(self):
+        A = self.A
+
+        a1 = A()
+        self.session.add(a1)
+        self.session.flush()
+
+        self._assert_count(1)
+        self.session.rollback()
+        self._assert_count(0)
+
+        a1 = A()
+        self.session.add(a1)
+        self.session.commit()
+        self._assert_count(1)
+
+        a2 = A()
+
+        self.session.add(a2)
+        self.session.flush()
+        self._assert_count(2)
+
+        self.session.rollback()
+        self._assert_count(1)
+
+    def _assert_count(self, count):
+        result = self.connection.scalar(
+            select(func.count()).select_from(self.table)
+        )
+        eq_(result, count)
+
+    def teardown(self):
+        self.teardown_session()
+
+        with self.connection.begin():
+            self._assert_count(0)
+
+        with self.connection.begin():
+            self.table.drop(self.connection)
+
+        # return connection to the Engine
+        self.connection.close()
+
+
+class NewStyleJoinIntoAnExternalTransactionTest(
+    JoinIntoAnExternalTransactionFixture
+):
+    """A new recipe for "join into an external transaction" that works
+    for both legacy and future engines/sessions
+
+    """
+
+    def setup_session(self):
+        # begin a non-ORM transaction
+        self.trans = self.connection.begin()
+
+        class A(object):
+            pass
+
+        mapper(A, self.table)
+        self.A = A
+
+        # bind an individual Session to the connection
+        self.session = Session(bind=self.connection, future=True)
+
+        if testing.requires.savepoints.enabled:
+            self.nested = self.connection.begin_nested()
+
+            @event.listens_for(self.session, "after_transaction_end")
+            def end_savepoint(session, transaction):
+                if not self.nested.is_active:
+                    self.nested = self.connection.begin_nested()
+
+    def teardown_session(self):
+        self.session.close()
+
+        # rollback - everything that happened with the
+        # Session above (including calls to commit())
+        # is rolled back.
+        self.trans.rollback()
+
+
+class FutureJoinIntoAnExternalTransactionTest(
+    NewStyleJoinIntoAnExternalTransactionTest,
+    fixtures.FutureEngineMixin,
+    fixtures.TestBase,
+):
+    pass
+
+
+class NonFutureJoinIntoAnExternalTransactionTest(
+    NewStyleJoinIntoAnExternalTransactionTest, fixtures.TestBase,
+):
+    pass
+
+
+class LegacyJoinIntoAnExternalTransactionTest(
+    JoinIntoAnExternalTransactionFixture, fixtures.TestBase,
+):
+    def setup_session(self):
+        # begin a non-ORM transaction
+        self.trans = self.connection.begin()
+
+        class A(object):
+            pass
+
+        mapper(A, self.table)
+        self.A = A
+
+        # bind an individual Session to the connection
+        self.session = Session(bind=self.connection)
+
+        if testing.requires.savepoints.enabled:
+            # start the session in a SAVEPOINT...
+            self.session.begin_nested()
+
+            # then each time that SAVEPOINT ends, reopen it
+            @event.listens_for(self.session, "after_transaction_end")
+            def restart_savepoint(session, transaction):
+                if transaction.nested and not transaction._parent.nested:
+
+                    # ensure that state is expired the way
+                    # session.commit() at the top level normally does
+                    # (optional step)
+                    session.expire_all()
+
+                    session.begin_nested()
+
+    def teardown_session(self):
+        self.session.close()
+
+        # rollback - everything that happened with the
+        # Session above (including calls to commit())
+        # is rolled back.
+        self.trans.rollback()
+
+
+class LegacyBranchedJoinIntoAnExternalTransactionTest(
+    LegacyJoinIntoAnExternalTransactionTest
+):
+    def setup_session(self):
+        # begin a non-ORM transaction
+        self.trans = self.connection.begin()
+
+        class A(object):
+            pass
+
+        mapper(A, self.table)
+        self.A = A
+
+        # neutron is doing this inside of a migration
+        # 1df244e556f5_add_unique_ha_router_agent_port_bindings.py
+        self.session = Session(bind=self.connection.connect())
+
+        if testing.requires.savepoints.enabled:
+            # start the session in a SAVEPOINT...
+            self.session.begin_nested()
+
+            # then each time that SAVEPOINT ends, reopen it
+            @event.listens_for(self.session, "after_transaction_end")
+            def restart_savepoint(session, transaction):
+                if transaction.nested and not transaction._parent.nested:
+
+                    # ensure that state is expired the way
+                    # session.commit() at the top level normally does
+                    # (optional step)
+                    session.expire_all()
+
+                    session.begin_nested()
diff --git a/test/orm/test_unitofwork.py b/test/orm/test_unitofwork.py
index 4f89711e7..e22f7beb3 100644
--- a/test/orm/test_unitofwork.py
+++ b/test/orm/test_unitofwork.py
@@ -754,12 +754,12 @@ class PassiveDeletesTest(fixtures.MappedTest):
         )
         mapper(MyClass, mytable)
 
-        session = create_session()
+        session = Session()
         mc = MyClass()
         mco = MyOtherClass()
         mco.myclass = mc
         session.add(mco)
-        session.flush()
+        session.commit()
 
         eq_(session.scalar(select(func.count("*")).select_from(mytable)), 1)
         eq_(
@@ -769,7 +769,7 @@ class PassiveDeletesTest(fixtures.MappedTest):
 
         session.expire(mco, ["myclass"])
         session.delete(mco)
-        session.flush()
+        session.commit()
 
         # mytable wasn't deleted, is the point.
         eq_(session.scalar(select(func.count("*")).select_from(mytable)), 1)
diff --git a/test/sql/test_case_statement.py b/test/sql/test_case_statement.py
index 491ff42bc..f2a88bd73 100644
--- a/test/sql/test_case_statement.py
+++ b/test/sql/test_case_statement.py
@@ -57,13 +57,8 @@ class CaseTest(fixtures.TestBase, AssertsCompiledSQL):
         inner = select(
             [
                 case(
-                    [
-                        [info_table.c.pk < 3, "lessthan3"],
-                        [
-                            and_(info_table.c.pk >= 3, info_table.c.pk < 7),
-                            "gt3",
-                        ],
-                    ]
+                    (info_table.c.pk < 3, "lessthan3"),
+                    (and_(info_table.c.pk >= 3, info_table.c.pk < 7), "gt3"),
                 ).label("x"),
                 info_table.c.pk,
                 info_table.c.info,
@@ -80,14 +75,17 @@ class CaseTest(fixtures.TestBase, AssertsCompiledSQL):
         # gt3 4 pk_4_data
         # gt3 5 pk_5_data
         # gt3 6 pk_6_data
-        assert inner_result == [
-            ("lessthan3", 1, "pk_1_data"),
-            ("lessthan3", 2, "pk_2_data"),
-            ("gt3", 3, "pk_3_data"),
-            ("gt3", 4, "pk_4_data"),
-            ("gt3", 5, "pk_5_data"),
-            ("gt3", 6, "pk_6_data"),
-        ]
+        eq_(
+            inner_result,
+            [
+                ("lessthan3", 1, "pk_1_data"),
+                ("lessthan3", 2, "pk_2_data"),
+                ("gt3", 3, "pk_3_data"),
+                ("gt3", 4, "pk_4_data"),
+                ("gt3", 5, "pk_5_data"),
+                ("gt3", 6, "pk_6_data"),
+            ],
+        )
 
         outer = select([inner.alias("q_inner")])
 
@@ -105,10 +103,8 @@ class CaseTest(fixtures.TestBase, AssertsCompiledSQL):
         w_else = select(
             [
                 case(
-                    [
-                        [info_table.c.pk < 3, cast(3, Integer)],
-                        [and_(info_table.c.pk >= 3, info_table.c.pk < 6), 6],
-                    ],
+                    [info_table.c.pk < 3, cast(3, Integer)],
+                    [and_(info_table.c.pk >= 3, info_table.c.pk < 6), 6],
                     else_=0,
                 ).label("x"),
                 info_table.c.pk,
@@ -119,21 +115,24 @@ class CaseTest(fixtures.TestBase, AssertsCompiledSQL):
 
         else_result = w_else.execute().fetchall()
 
-        assert else_result == [
-            (3, 1, "pk_1_data"),
-            (3, 2, "pk_2_data"),
-            (6, 3, "pk_3_data"),
-            (6, 4, "pk_4_data"),
-            (6, 5, "pk_5_data"),
-            (0, 6, "pk_6_data"),
-        ]
+        eq_(
+            else_result,
+            [
+                (3, 1, "pk_1_data"),
+                (3, 2, "pk_2_data"),
+                (6, 3, "pk_3_data"),
+                (6, 4, "pk_4_data"),
+                (6, 5, "pk_5_data"),
+                (0, 6, "pk_6_data"),
+            ],
+        )
 
     def test_literal_interpretation_ambiguous(self):
         assert_raises_message(
             exc.ArgumentError,
             r"Column expression expected, got 'x'",
             case,
-            [("x", "y")],
+            ("x", "y"),
         )
 
     def test_literal_interpretation_ambiguous_tuple(self):
@@ -141,18 +140,18 @@ class CaseTest(fixtures.TestBase, AssertsCompiledSQL):
             exc.ArgumentError,
             r"Column expression expected, got \('x', 'y'\)",
             case,
-            [(("x", "y"), "z")],
+            (("x", "y"), "z"),
         )
 
     def test_literal_interpretation(self):
         t = table("test", column("col1"))
 
         self.assert_compile(
-            case([("x", "y")], value=t.c.col1),
+            case(("x", "y"), value=t.c.col1),
             "CASE test.col1 WHEN :param_1 THEN :param_2 END",
         )
         self.assert_compile(
-            case([(t.c.col1 == 7, "y")], else_="z"),
+            case((t.c.col1 == 7, "y"), else_="z"),
             "CASE WHEN (test.col1 = :col1_1) THEN :param_1 ELSE :param_2 END",
         )
 
@@ -162,7 +161,7 @@ class CaseTest(fixtures.TestBase, AssertsCompiledSQL):
             select(
                 [
                     case(
-                        [(info_table.c.info == "pk_4_data", text("'yes'"))],
+                        (info_table.c.info == "pk_4_data", text("'yes'")),
                         else_=text("'no'"),
                     )
                 ]
@@ -170,36 +169,20 @@ class CaseTest(fixtures.TestBase, AssertsCompiledSQL):
             select(
                 [
                     case(
-                        [
-                            (
-                                info_table.c.info == "pk_4_data",
-                                literal_column("'yes'"),
-                            )
-                        ],
+                        (
+                            info_table.c.info == "pk_4_data",
+                            literal_column("'yes'"),
+                        ),
                         else_=literal_column("'no'"),
                     )
                 ]
             ).order_by(info_table.c.info),
         ]:
-            if testing.against("firebird"):
-                eq_(
-                    s.execute().fetchall(),
-                    [
-                        ("no ",),
-                        ("no ",),
-                        ("no ",),
-                        ("yes",),
-                        ("no ",),
-                        ("no ",),
-                    ],
-                )
-            else:
-                eq_(
-                    s.execute().fetchall(),
-                    [("no",), ("no",), ("no",), ("yes",), ("no",), ("no",)],
-                )
+            eq_(
+                s.execute().fetchall(),
+                [("no",), ("no",), ("no",), ("yes",), ("no",), ("no",)],
+            )
 
-    @testing.fails_on("firebird", "FIXME: unknown")
     def testcase_with_dict(self):
         query = select(
             [
@@ -215,24 +198,27 @@ class CaseTest(fixtures.TestBase, AssertsCompiledSQL):
             ],
             from_obj=[info_table],
         )
-        assert query.execute().fetchall() == [
-            ("lessthan3", 1, "pk_1_data"),
-            ("lessthan3", 2, "pk_2_data"),
-            ("gt3", 3, "pk_3_data"),
-            ("gt3", 4, "pk_4_data"),
-            ("gt3", 5, "pk_5_data"),
-            ("gt3", 6, "pk_6_data"),
-        ]
-
-        simple_query = select(
+        eq_(
+            query.execute().fetchall(),
             [
+                ("lessthan3", 1, "pk_1_data"),
+                ("lessthan3", 2, "pk_2_data"),
+                ("gt3", 3, "pk_3_data"),
+                ("gt3", 4, "pk_4_data"),
+                ("gt3", 5, "pk_5_data"),
+                ("gt3", 6, "pk_6_data"),
+            ],
+        )
+
+        simple_query = (
+            select(
                 case(
                     {1: "one", 2: "two"}, value=info_table.c.pk, else_="other"
                 ),
                 info_table.c.pk,
-            ],
-            whereclause=info_table.c.pk < 4,
-            from_obj=[info_table],
+            )
+            .where(info_table.c.pk < 4)
+            .select_from(info_table)
         )
 
         assert simple_query.execute().fetchall() == [
diff --git a/test/sql/test_compare.py b/test/sql/test_compare.py
index fff5171ef..7ac716dbe 100644
--- a/test/sql/test_compare.py
+++ b/test/sql/test_compare.py
@@ -320,20 +320,15 @@ class CoreFixtures(object):
             ClauseList(table_a.c.a == 5, table_a.c.b == table_a.c.a),
         ),
         lambda: (
-            case(whens=[(table_a.c.a == 5, 10), (table_a.c.a == 10, 20)]),
-            case(whens=[(table_a.c.a == 18, 10), (table_a.c.a == 10, 20)]),
-            case(whens=[(table_a.c.a == 5, 10), (table_a.c.b == 10, 20)]),
+            case((table_a.c.a == 5, 10), (table_a.c.a == 10, 20)),
+            case((table_a.c.a == 18, 10), (table_a.c.a == 10, 20)),
+            case((table_a.c.a == 5, 10), (table_a.c.b == 10, 20)),
             case(
-                whens=[
-                    (table_a.c.a == 5, 10),
-                    (table_a.c.b == 10, 20),
-                    (table_a.c.a == 9, 12),
-                ]
-            ),
-            case(
-                whens=[(table_a.c.a == 5, 10), (table_a.c.a == 10, 20)],
-                else_=30,
+                (table_a.c.a == 5, 10),
+                (table_a.c.b == 10, 20),
+                (table_a.c.a == 9, 12),
             ),
+            case((table_a.c.a == 5, 10), (table_a.c.a == 10, 20), else_=30,),
             case({"wendy": "W", "jack": "J"}, value=table_a.c.a, else_="E"),
             case({"wendy": "W", "jack": "J"}, value=table_a.c.b, else_="E"),
             case({"wendy_w": "W", "jack": "J"}, value=table_a.c.a, else_="E"),
diff --git a/test/sql/test_compiler.py b/test/sql/test_compiler.py
index 7f06aa0d1..d79d00555 100644
--- a/test/sql/test_compiler.py
+++ b/test/sql/test_compiler.py
@@ -4050,7 +4050,7 @@ class KwargPropagationTest(fixtures.TestBase):
         self._do_test(s)
 
     def test_case(self):
-        c = case([(self.criterion, self.column)], else_=self.column)
+        c = case((self.criterion, self.column), else_=self.column)
         self._do_test(c)
 
     def test_cast(self):
diff --git a/test/sql/test_deprecations.py b/test/sql/test_deprecations.py
index e68b1398b..871d09e04 100644
--- a/test/sql/test_deprecations.py
+++ b/test/sql/test_deprecations.py
@@ -3,6 +3,7 @@
 from sqlalchemy import alias
 from sqlalchemy import and_
 from sqlalchemy import bindparam
+from sqlalchemy import case
 from sqlalchemy import CHAR
 from sqlalchemy import column
 from sqlalchemy import create_engine
@@ -488,6 +489,111 @@ class SelectableTest(fixtures.TestBase, AssertsCompiledSQL):
             "SELECT anon_1.a FROM (SELECT 1 AS a ORDER BY 1) AS anon_1",
         )
 
+    def test_case_list_legacy(self):
+        t1 = table("t", column("q"))
+
+        with testing.expect_deprecated(
+            r"The \"whens\" argument to case\(\) is now passed"
+        ):
+            stmt = select(t1).where(
+                case(
+                    [(t1.c.q == 5, "foo"), (t1.c.q == 10, "bar")], else_="bat"
+                )
+                != "bat"
+            )
+
+        self.assert_compile(
+            stmt,
+            "SELECT t.q FROM t WHERE CASE WHEN (t.q = :q_1) "
+            "THEN :param_1 WHEN (t.q = :q_2) THEN :param_2 "
+            "ELSE :param_3 END != :param_4",
+        )
+
+    def test_case_whens_kw(self):
+        t1 = table("t", column("q"))
+
+        with testing.expect_deprecated(
+            r"The \"whens\" argument to case\(\) is now passed"
+        ):
+            stmt = select(t1).where(
+                case(
+                    whens=[(t1.c.q == 5, "foo"), (t1.c.q == 10, "bar")],
+                    else_="bat",
+                )
+                != "bat"
+            )
+
+        self.assert_compile(
+            stmt,
+            "SELECT t.q FROM t WHERE CASE WHEN (t.q = :q_1) "
+            "THEN :param_1 WHEN (t.q = :q_2) THEN :param_2 "
+            "ELSE :param_3 END != :param_4",
+        )
+
+    def test_case_whens_dict_kw(self):
+        t1 = table("t", column("q"))
+
+        with testing.expect_deprecated(
+            r"The \"whens\" argument to case\(\) is now passed"
+        ):
+            stmt = select(t1).where(
+                case(whens={t1.c.q == 5: "foo"}, else_="bat",) != "bat"
+            )
+
+        self.assert_compile(
+            stmt,
+            "SELECT t.q FROM t WHERE CASE WHEN (t.q = :q_1) THEN "
+            ":param_1 ELSE :param_2 END != :param_3",
+        )
+
+    def test_case_kw_arg_detection(self):
+        # because we support py2k, case() has to parse **kw for now
+
+        assert_raises_message(
+            TypeError,
+            "unknown arguments: bat, foo",
+            case,
+            (column("x") == 10, 5),
+            else_=15,
+            foo="bar",
+            bat="hoho",
+        )
+
+    def test_with_only_generative(self):
+        table1 = table(
+            "table1",
+            column("col1"),
+            column("col2"),
+            column("col3"),
+            column("colx"),
+        )
+        s1 = table1.select().scalar_subquery()
+
+        with testing.expect_deprecated(
+            r"The \"columns\" argument to "
+            r"Select.with_only_columns\(\) is now passed"
+        ):
+            stmt = s1.with_only_columns([s1])
+        self.assert_compile(
+            stmt,
+            "SELECT (SELECT table1.col1, table1.col2, "
+            "table1.col3, table1.colx FROM table1) AS anon_1",
+        )
+
+    def test_from_list_with_columns(self):
+        table1 = table("t1", column("a"))
+        table2 = table("t2", column("b"))
+        s1 = select(table1.c.a, table2.c.b)
+        self.assert_compile(s1, "SELECT t1.a, t2.b FROM t1, t2")
+
+        with testing.expect_deprecated(
+            r"The \"columns\" argument to "
+            r"Select.with_only_columns\(\) is now passed"
+        ):
+            s2 = s1.with_only_columns([table2.c.b])
+
+        self.assert_compile(s2, "SELECT t2.b FROM t2")
+
     def test_column(self):
         stmt = select(column("x"))
         with testing.expect_deprecated(
@@ -815,7 +921,7 @@ class DeprecatedAppendMethTest(fixtures.TestBase, AssertsCompiledSQL):
     def test_append_column(self):
         t1 = table("t1", column("q"), column("p"))
         stmt = select(t1.c.q)
-        with self._expect_deprecated("Select", "column", "column"):
+        with self._expect_deprecated("Select", "column", "add_columns"):
             stmt.append_column(t1.c.p)
         self.assert_compile(stmt, "SELECT t1.q, t1.p FROM t1")
 
diff --git a/test/sql/test_selectable.py b/test/sql/test_selectable.py
index 55875632a..01c8d7ca6 100644
--- a/test/sql/test_selectable.py
+++ b/test/sql/test_selectable.py
@@ -479,7 +479,7 @@ class SelectableTest(
     def test_with_only_generative(self):
         s1 = table1.select().scalar_subquery()
         self.assert_compile(
-            s1.with_only_columns([s1]),
+            s1.with_only_columns(s1),
             "SELECT (SELECT table1.col1, table1.col2, "
             "table1.col3, table1.colx FROM table1) AS anon_1",
         )
@@ -1165,12 +1165,12 @@ class SelectableTest(
         table2 = table("t2", column("b"))
         s1 = select(table1.c.a, table2.c.b)
         self.assert_compile(s1, "SELECT t1.a, t2.b FROM t1, t2")
-        s2 = s1.with_only_columns([table2.c.b])
+        s2 = s1.with_only_columns(table2.c.b)
         self.assert_compile(s2, "SELECT t2.b FROM t2")
 
         s3 = sql_util.ClauseAdapter(table1).traverse(s1)
         self.assert_compile(s3, "SELECT t1.a, t2.b FROM t1, t2")
-        s4 = s3.with_only_columns([table2.c.b])
+        s4 = s3.with_only_columns(table2.c.b)
         self.assert_compile(s4, "SELECT t2.b FROM t2")
 
     def test_from_list_against_existing_one(self):