Don't assume m2o key is present in the dictionary

Fixed regression caused by new selectinload for many-to-one logic where
a primaryjoin condition not based on real foreign keys would cause
KeyError if a related object did not exist for a given key value on the
parent object.

Fixes: #4777
Change-Id: I4ba96318302be68abe0e8c3611684087a04a2322
(cherry picked from commit 8f5b65f431)

doc/build/changelog/unreleased_13/4777.rst

@@ -0,0 +1,8 @@
+.. change::
+    :tags: bug, orm
+    :tickets: 4777
+
+    Fixed regression caused by new selectinload for many-to-one logic where
+    a primaryjoin condition not based on real foreign keys would cause
+    KeyError if a related object did not exist for a given key value on the
+    parent object.

lib/sqlalchemy/orm/strategies.py

@@ -2381,7 +2381,12 @@ class SelectInLoader(AbstractRelationshipLoader, util.MemoizedSlots):
             }
 
             for key in chunk:
-                related_obj = data[key]
+                # for a real foreign key and no concurrent changes to the
+                # DB while running this method, "key" is always present in
+                # data.  However, for primaryjoins without real foreign keys
+                # a non-None primaryjoin condition may still refer to no
+                # related object.
+                related_obj = data.get(key, None)
                 for state, dict_, overwrite in our_states[key]:
                     if not overwrite and self.key in dict_:
                         continue
@@ -2431,6 +2436,8 @@ class SelectInLoader(AbstractRelationshipLoader, util.MemoizedSlots):
                         state, state_dict, collection[0]
                     )
                 else:
+                    # note that empty tuple set on uselist=False sets the
+                    # value to None
                     state.get_impl(self.key).set_committed_value(
                         state, state_dict, collection
                     )

test/orm/test_selectin_relations.py

@@ -1120,6 +1120,31 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL):
 
         self.assert_sql_count(testing.db, go, 2)
 
+    def test_one_to_many_scalar_none(self):
+        Address, addresses, users, User = (
+            self.classes.Address,
+            self.tables.addresses,
+            self.tables.users,
+            self.classes.User,
+        )
+
+        mapper(
+            User,
+            users,
+            properties=dict(
+                address=relationship(
+                    mapper(Address, addresses), lazy="selectin", uselist=False
+                )
+            ),
+        )
+        q = create_session().query(User)
+
+        def go():
+            result = q.filter(users.c.id == 10).all()
+            eq_([User(id=10, address=None)], result)
+
+        self.assert_sql_count(testing.db, go, 2)
+
     def test_many_to_one(self):
         users, Address, addresses, User = (
             self.tables.users,
@@ -2937,6 +2962,59 @@ class SingleInhSubclassTest(
         )
 
 
+class MissingForeignTest(
+    fixtures.DeclarativeMappedTest, testing.AssertsExecutionResults
+):
+    @classmethod
+    def setup_classes(cls):
+        Base = cls.DeclarativeBasic
+
+        class A(fixtures.ComparableEntity, Base):
+            __tablename__ = "a"
+            id = Column(Integer, primary_key=True)
+            b_id = Column(Integer)
+            b = relationship("B", primaryjoin="foreign(A.b_id) == B.id")
+            q = Column(Integer)
+
+        class B(fixtures.ComparableEntity, Base):
+            __tablename__ = "b"
+            id = Column(Integer, primary_key=True)
+            x = Column(Integer)
+            y = Column(Integer)
+
+    @classmethod
+    def insert_data(cls):
+        A, B = cls.classes("A", "B")
+
+        s = Session()
+        b1, b2 = B(id=1, x=5, y=9), B(id=2, x=10, y=8)
+        s.add_all(
+            [
+                A(id=1, b_id=1),
+                A(id=2, b_id=5),
+                A(id=3, b_id=2),
+                A(id=4, b=None),
+                b1,
+                b2,
+            ]
+        )
+        s.commit()
+
+    def test_missing_rec(self):
+        A, B = self.classes("A", "B")
+
+        s = Session()
+        eq_(
+            s.query(A).options(selectinload(A.b)).order_by(A.id).all(),
+            [
+                A(id=1, b=B(id=1)),
+                A(id=2, b=None, b_id=5),
+                A(id=3, b=B(id=2)),
+                A(id=4, b=None, b_id=None),
+            ],
+        )
+
+
 class M2OWDegradeTest(
     fixtures.DeclarativeMappedTest, testing.AssertsExecutionResults
 ):