fix: Session.get() with with_for_update=False skips identity map

Fixes #13176.

`Session.get()` checks `with_for_update is None` to decide whether to look up the identity map. Passing `with_for_update=False` fails this check and always hits the database, even though `ForUpdateArg._from_argument` already treats `False` and `None` identically (both return `None`). Changed to `with_for_update in (None, False)` to match.

Closes: #13199
Pull-request: https://github.com/sqlalchemy/sqlalchemy/pull/13199
Pull-request-sha: c639b8723888947317598df387c5e3e4d87acac4

Change-Id: I0584873f46099afadcdd760c0a267ae4d30528eb

diff --git a/doc/build/changelog/unreleased_20/13176.rst b/doc/build/changelog/unreleased_20/13176.rst
new file mode 100644 (file)
index 0000000..78dfb43
--- /dev/null
+++ b/doc/build/changelog/unreleased_20/13176.rst
@@ -0,0 +1,8 @@
+.. change::
+    :tags: bug, orm
+    :tickets: 13176
+
+    Fixed issue where :meth:`_orm.Session.get` would bypass the identity map
+    and emit unnecessary SQL when ``with_for_update=False`` was passed,
+    rather than treating it equivalently to the default of ``None``.
+    Pull request courtesy of Joshua Swanson.

diff --git a/lib/sqlalchemy/orm/session.py b/lib/sqlalchemy/orm/session.py
index b33a6ba651578bf40c184fc3958fb8f80f3aadf1..0aa6458f97dff8f41d3fca87c4e30f367ac1dbd7 100644 (file)
--- a/lib/sqlalchemy/orm/session.py
+++ b/lib/sqlalchemy/orm/session.py
@@ -3904,10 +3904,12 @@ class Session(_SessionClassMethods, EventTarget):
                     )
                 ) from err
 
+        for_update_arg = ForUpdateArg._from_argument(with_for_update)
+
         if (
             not populate_existing
             and not mapper.always_refresh
-            and with_for_update is None
+            and for_update_arg is None
         ):
             instance = self._identity_lookup(
                 mapper,
@@ -3932,10 +3934,8 @@ class Session(_SessionClassMethods, EventTarget):
         if populate_existing:
             load_options += {"_populate_existing": populate_existing}
         statement = sql.select(mapper)
-        if with_for_update is not None:
-            statement._for_update_arg = ForUpdateArg._from_argument(
-                with_for_update
-            )
+        if for_update_arg is not None:
+            statement._for_update_arg = for_update_arg
 
         if options:
             statement = statement.options(*options)

diff --git a/test/orm/test_session.py b/test/orm/test_session.py
index 349768c21b2e8bd71852369d0854307da883fb47..fd14f5acdabf4c7f568b0d07ab848c6ac07939e4 100644 (file)
--- a/test/orm/test_session.py
+++ b/test/orm/test_session.py
@@ -642,6 +642,48 @@ class SessionUtilTest(_fixtures.FixtureTest):
         u2 = s.get(User, 7)
         is_not(u, u2)
 
+    @testing.variation(
+        "with_for_update_arg", ["true", "false", "none", "omitted"]
+    )
+    def test_get_with_for_update_use(self, with_for_update_arg):
+        """test #13176"""
+        users, User = self.tables.users, self.classes.User
+        self.mapper_registry.map_imperatively(User, users)
+
+        s = fixture_session()
+        s.execute(
+            insert(self.tables.users),
+            [{"id": 7, "name": "7"}],
+        )
+        u = s.get(User, 7)
+
+        if with_for_update_arg.true:
+            self.assert_sql_count(
+                testing.db,
+                lambda: s.get(User, 7, with_for_update=True),
+                1,
+            )
+        elif with_for_update_arg.false:
+            self.assert_sql_count(
+                testing.db,
+                lambda: is_(s.get(User, 7, with_for_update=False), u),
+                0,
+            )
+        elif with_for_update_arg.none:
+            self.assert_sql_count(
+                testing.db,
+                lambda: is_(s.get(User, 7, with_for_update=None), u),
+                0,
+            )
+        elif with_for_update_arg.omitted:
+            self.assert_sql_count(
+                testing.db,
+                lambda: is_(s.get(User, 7), u),
+                0,
+            )
+        else:
+            with_for_update_arg.fail()
+
     def test_get_one(self):
         users, User = self.tables.users, self.classes.User
         self.mapper_registry.map_imperatively(User, users)