]> git.ipfire.org Git - thirdparty/sqlalchemy/sqlalchemy.git/commitdiff
resolve synonyms in dictionary form of Session.get()
authorMike Bayer <mike_mp@zzzcomputing.com>
Fri, 4 Nov 2022 16:48:43 +0000 (12:48 -0400)
committerMike Bayer <mike_mp@zzzcomputing.com>
Fri, 4 Nov 2022 16:49:49 +0000 (12:49 -0400)
Improved "dictionary mode" for :meth:`_orm.Session.get` so that synonym
names which refer to primary key attribute names may be indicated in the
named dictionary.

Fixes: #8753
Change-Id: I56112564a5c23b51b26e01c64087cbf4399cd951

doc/build/changelog/unreleased_14/8753.rst [new file with mode: 0644]
lib/sqlalchemy/orm/mapper.py
lib/sqlalchemy/orm/session.py
test/orm/test_query.py

diff --git a/doc/build/changelog/unreleased_14/8753.rst b/doc/build/changelog/unreleased_14/8753.rst
new file mode 100644 (file)
index 0000000..6f898e9
--- /dev/null
@@ -0,0 +1,7 @@
+.. change::
+    :tags: bug, orm
+    :tickets: 8753
+
+    Improved "dictionary mode" for :meth:`_orm.Session.get` so that synonym
+    names which refer to primary key attribute names may be indicated in the
+    named dictionary.
index 89072873838099ee21f6b2260570292ab9d19a86..5f7ff43e42001f7b14cccfef558d7c30cc5af2bf 100644 (file)
@@ -2908,6 +2908,24 @@ class Mapper(
             dict(self.class_manager._all_sqla_attributes())
         )
 
+    @HasMemoized.memoized_attribute
+    @util.preload_module("sqlalchemy.orm.descriptor_props")
+    def _pk_synonyms(self) -> Dict[str, str]:
+        """return a dictionary of {syn_attribute_name: pk_attr_name} for
+        all synonyms that refer to primary key columns
+
+        """
+        descriptor_props = util.preloaded.orm_descriptor_props
+
+        pk_keys = {prop.key for prop in self._identity_key_props}
+
+        return {
+            syn.key: syn.name
+            for k, syn in self._props.items()
+            if isinstance(syn, descriptor_props.SynonymProperty)
+            and syn.name in pk_keys
+        }
+
     @HasMemoized.memoized_attribute
     @util.preload_module("sqlalchemy.orm.descriptor_props")
     def synonyms(self) -> util.ReadOnlyProperties[SynonymProperty[Any]]:
index 540e64e7d38605e5fbbb3accddb0eadd50a8ba8d..849d37ab227375e73627396bd48c108f932f2b17 100644 (file)
@@ -3399,6 +3399,21 @@ class Session(_SessionClassMethods, EventTarget):
             )
 
         if is_dict:
+
+            pk_synonyms = mapper._pk_synonyms
+
+            if pk_synonyms:
+                correct_keys = set(pk_synonyms).intersection(
+                    primary_key_identity
+                )
+
+                if correct_keys:
+                    primary_key_identity = dict(primary_key_identity)
+                    for k in correct_keys:
+                        primary_key_identity[
+                            pk_synonyms[k]
+                        ] = primary_key_identity[k]
+
             try:
                 primary_key_identity = list(
                     primary_key_identity[prop.key]
@@ -3409,7 +3424,7 @@ class Session(_SessionClassMethods, EventTarget):
                 raise sa_exc.InvalidRequestError(
                     "Incorrect names of values in identifier to formulate "
                     "primary key for session.get(); primary key attribute "
-                    "names are %s"
+                    "names are %s (synonym names are also accepted)"
                     % ",".join(
                         "'%s'" % prop.key
                         for prop in mapper._identity_key_props
index c05fdaf4fc456b07c175e4e281556b55eefc8748..dcf7735cd5febbb763df60aa9fcbf948e127d595 100644 (file)
@@ -1226,6 +1226,31 @@ class GetTest(QueryTest):
         u2 = s.get(User, 7)
         assert u is not u2
 
+    def test_get_synonym_direct_name(self, decl_base):
+        """test #8753"""
+
+        class MyUser(decl_base):
+            __table__ = self.tables.users
+
+            syn_id = synonym("id")
+
+        s = fixture_session()
+        u = s.get(MyUser, {"syn_id": 7})
+        eq_(u.id, 7)
+
+    def test_get_synonym_indirect(self, decl_base):
+        """test #8753"""
+
+        class MyUser(decl_base):
+            __table__ = self.tables.users
+
+            uid = __table__.c.id
+            syn_id = synonym("uid")
+
+        s = fixture_session()
+        u = s.get(MyUser, {"syn_id": 7})
+        eq_(u.uid, 7)
+
     def test_get_composite_pk_no_result(self):
         CompositePk = self.classes.CompositePk