Don't null FK for collection-removed item with passive_deletes='all'

Fixed issue regarding passive_deletes="all", where the foreign key
attribute of an object is maintained with its value even after the object
is removed from its parent collection.  Previously, the unit of work would
set this to NULL even though passive_deletes indicated it should not be
modified.

Change-Id: I5ba98bc388cbdd6323d255b764e02506c2e66896
Fixes: #3844

4 files changed, 77 insertions, 12 deletions

diff --git a/doc/build/changelog/migration_13.rst b/doc/build/changelog/migration_13.rst
index 86187a2f4..8cb1fd49c 100644
--- a/doc/build/changelog/migration_13.rst
+++ b/doc/build/changelog/migration_13.rst
@@ -87,6 +87,47 @@ and can't easily be generalized for more complex queries.
 
 :ticket:`4246`
 
+.. _change_3844
+
+passive_deletes='all' will leave FK unchanged for object removed from collection
+--------------------------------------------------------------------------------
+
+The :paramref:`.relationship.passive_deletes` option accepts the value
+``"all"`` to indicate that no foreign key attributes should be modified when
+the object is flushed, even if the relationship's collection / reference has
+been removed.   Previously, this did not take place for one-to-many, or
+one-to-one relationships, in the following situation::
+
+    class User(Base):
+        __tablename__ = 'users'
+
+        id = Column(Integer, primary_key=True)
+        addresses = relationship(
+            "Address",
+            passive_deletes="all")
+
+    class Address(Base):
+        __tablename__ = 'addresses'
+        id = Column(Integer, primary_key=True)
+        email = Column(String)
+
+        user_id = Column(Integer, ForeignKey('users.id'))
+        user = relationship("User")
+
+    u1 = session.query(User).first()
+    address = u1.addresses[0]
+    u1.addresses.remove(address)
+    session.commit()
+
+    # would fail and be set to None
+    assert address.user_id == u1.id
+
+The fix now includes that ``address.user_id`` is left unchanged as per
+``passive_deletes="all"``. This kind of thing is useful for building custom
+"version table" schemes and such where rows are archived instead of deleted.
+
+:ticket:`3844`
+
 New Features and Improvements - Core
 ====================================
 
diff --git a/doc/build/changelog/unreleased_13/3844.rst b/doc/build/changelog/unreleased_13/3844.rst
new file mode 100644
index 000000000..8c65c47cd
--- /dev/null
+++ b/doc/build/changelog/unreleased_13/3844.rst
@@ -0,0 +1,13 @@
+.. change::
+    :tags: bug, orm
+    :tickets: 3844
+
+    Fixed issue regarding passive_deletes="all", where the foreign key
+    attribute of an object is maintained with its value even after the object
+    is removed from its parent collection.  Previously, the unit of work would
+    set this to NULL even though passive_deletes indicated it should not be
+    modified.
+
+    .. seealso::
+
+        :ref:`change_3844`
diff --git a/lib/sqlalchemy/orm/dependency.py b/lib/sqlalchemy/orm/dependency.py
index 799e633b3..1a68ea9c7 100644
--- a/lib/sqlalchemy/orm/dependency.py
+++ b/lib/sqlalchemy/orm/dependency.py
@@ -434,6 +434,9 @@ class OneToManyDP(DependencyProcessor):
     def presort_saves(self, uowcommit, states):
         children_added = uowcommit.memo(('children_added', self), set)
 
+        should_null_fks = not self.cascade.delete_orphan and \
+            not self.passive_deletes == 'all'
+
         for state in states:
             pks_changed = self._pks_changed(uowcommit, state)
 
@@ -457,9 +460,10 @@ class OneToManyDP(DependencyProcessor):
 
                 for child in history.deleted:
                     if not self.cascade.delete_orphan:
-                        uowcommit.register_object(child, isdelete=False,
-                                                  operation='delete',
-                                                  prop=self.prop)
+                        if should_null_fks:
+                            uowcommit.register_object(child, isdelete=False,
+                                                      operation='delete',
+                                                      prop=self.prop)
                     elif self.hasparent(child) is False:
                         uowcommit.register_object(
                             child, isdelete=True,
@@ -528,6 +532,9 @@ class OneToManyDP(DependencyProcessor):
                     # if the old parent wasn't deleted but child was moved.
 
     def process_saves(self, uowcommit, states):
+        should_null_fks = not self.cascade.delete_orphan and \
+            not self.passive_deletes == 'all'
+
         for state in states:
             history = uowcommit.get_attribute_history(
                 state,
@@ -541,7 +548,7 @@ class OneToManyDP(DependencyProcessor):
                         self._post_update(child, uowcommit, [state])
 
                 for child in history.deleted:
-                    if not self.cascade.delete_orphan and \
+                    if should_null_fks and not self.cascade.delete_orphan and \
                             not self.hasparent(child):
                         self._synchronize(state, child, None, True,
                                           uowcommit, False)
diff --git a/test/orm/test_unitofwork.py b/test/orm/test_unitofwork.py
index 8a4091ed4..94b62eb5b 100644
--- a/test/orm/test_unitofwork.py
+++ b/test/orm/test_unitofwork.py
@@ -747,8 +747,7 @@ class ExtraPassiveDeletesTest(fixtures.MappedTest):
         mc.children[0].data = 'some new data'
         assert_raises(sa.exc.DBAPIError, session.flush)
 
-    def test_extra_passive_obj_removed_o2m_still_nulls_out(self):
-        # see #3844, which we decided was not a bug
+    def test_extra_passive_obj_removed_o2m(self):
         myothertable, MyClass, MyOtherClass, mytable = (
             self.tables.myothertable,
             self.classes.MyClass,
@@ -758,19 +757,24 @@ class ExtraPassiveDeletesTest(fixtures.MappedTest):
         mapper(MyOtherClass, myothertable)
         mapper(MyClass, mytable, properties={
             'children': relationship(MyOtherClass,
-                                  passive_deletes='all')})
+                                     passive_deletes='all')})
 
         session = create_session()
         mc = MyClass()
-        moc = MyOtherClass()
-        mc.children.append(moc)
-        session.add_all([mc, moc])
+        moc1 = MyOtherClass()
+        moc2 = MyOtherClass()
+        mc.children.append(moc1)
+        mc.children.append(moc2)
+        session.add_all([mc, moc1, moc2])
         session.flush()
 
-        mc.children.remove(moc)
+        mc.children.remove(moc1)
+        mc.children.remove(moc2)
+        moc1.data = 'foo'
         session.flush()
 
-        eq_(moc.parent_id, None)
+        eq_(moc1.parent_id, mc.id)
+        eq_(moc2.parent_id, mc.id)
 
     def test_dont_emit(self):
         myothertable, MyClass, MyOtherClass, mytable = (