Add new NO_RAISE attribute flag and specify for m2o history load

Added new behavior to the lazy load that takes place when the "old" value of
a many-to-one is retrieved, such that exceptions which would be raised due
to either ``lazy="raise"`` or a detached session error are skipped.

Fixes: #4353
Change-Id: I6c6c77613e93061a909f5062b70b17e8913fc9ee

6 files changed, 280 insertions, 5 deletions

diff --git a/doc/build/changelog/migration_13.rst b/doc/build/changelog/migration_13.rst
index fee5ad5e9..eab13bef9 100644
--- a/doc/build/changelog/migration_13.rst
+++ b/doc/build/changelog/migration_13.rst
@@ -190,6 +190,41 @@ to ``None``::
 
 :ticket:`4308`
 
+.. _change_4353:
+
+Many-to-one replacement won't raise for "raiseload" or detached for "old" object
+--------------------------------------------------------------------------------
+
+Given the case where a lazy load would proceed on a many-to-one relationship
+in order to load the "old" value, if the relationship does not specify
+the :paramref:`.relationship.active_history` flag, an assertion will not
+be raised for a detached object::
+
+    a1 = session.query(Address).filter_by(id=5).one()
+
+    session.expunge(a1)
+
+    a1.user = some_user
+
+Above, when the ``.user`` attribute is replaced on the detached ``a1`` object,
+a :class:`.DetachedInstanceError` would be raised as the attribute is attempting
+to retrieve the previous value of ``.user`` from the identity map.  The change
+is that the operation now proceeds without the old value being loaded.
+
+The same change is also made to the ``lazy="raise"`` loader strategy::
+
+    class Address(Base):
+        # ...
+
+        user = relationship("User", ..., lazy="raise")
+
+Previously, the association of ``a1.user`` would invoke the "raiseload"
+exception as a result of the attribute attempting to retrieve the previous
+value.   This assertion is now skipped in the case of loading the "old" value.
+
+
+:ticket:`4353`
+
 .. _change_3423:
 
 AssociationProxy stores class-specific state in a separate container
diff --git a/doc/build/changelog/unreleased_13/4353.rst b/doc/build/changelog/unreleased_13/4353.rst
new file mode 100644
index 000000000..d94acf828
--- /dev/null
+++ b/doc/build/changelog/unreleased_13/4353.rst
@@ -0,0 +1,11 @@
+.. change::
+   :tags: bug, orm
+   :tickets: 4353
+
+   Added new behavior to the lazy load that takes place when the "old" value of
+   a many-to-one is retrieved, such that exceptions which would be raised due
+   to either ``lazy="raise"`` or a detached session error are skipped.
+
+   .. seealso::
+
+    :ref:`change_4353`
diff --git a/lib/sqlalchemy/orm/attributes.py b/lib/sqlalchemy/orm/attributes.py
index 3eaa41e8f..ff730d745 100644
--- a/lib/sqlalchemy/orm/attributes.py
+++ b/lib/sqlalchemy/orm/attributes.py
@@ -24,7 +24,8 @@ from .base import PASSIVE_NO_RESULT, ATTR_WAS_SET, ATTR_EMPTY, NO_VALUE,\
     NEVER_SET, NO_CHANGE, CALLABLES_OK, SQL_OK, RELATED_OBJECT_OK,\
     INIT_OK, NON_PERSISTENT_OK, LOAD_AGAINST_COMMITTED, PASSIVE_OFF,\
     PASSIVE_RETURN_NEVER_SET, PASSIVE_NO_INITIALIZE, PASSIVE_NO_FETCH,\
-    PASSIVE_NO_FETCH_RELATED, PASSIVE_ONLY_PERSISTENT, NO_AUTOFLUSH
+    PASSIVE_NO_FETCH_RELATED, PASSIVE_ONLY_PERSISTENT, NO_AUTOFLUSH, \
+    NO_RAISE
 from .base import state_str, instance_str
 
 
@@ -753,7 +754,7 @@ class ScalarObjectAttributeImpl(ScalarAttributeImpl):
         else:
             old = self.get(
                 state, dict_, passive=PASSIVE_NO_FETCH ^ INIT_OK |
-                LOAD_AGAINST_COMMITTED)
+                LOAD_AGAINST_COMMITTED | NO_RAISE)
 
         self.fire_remove_event(state, dict_, old, self._remove_token)
 
@@ -817,7 +818,7 @@ class ScalarObjectAttributeImpl(ScalarAttributeImpl):
         else:
             old = self.get(
                 state, dict_, passive=PASSIVE_NO_FETCH ^ INIT_OK |
-                LOAD_AGAINST_COMMITTED)
+                LOAD_AGAINST_COMMITTED | NO_RAISE)
 
         if check_old is not None and \
                 old is not PASSIVE_NO_RESULT and \
@@ -1253,7 +1254,9 @@ def backref_listeners(attribute, key, uselist):
         return child
 
     def emit_backref_from_collection_remove_event(state, child, initiator):
-        if child is not None:
+        if child is not None and \
+                child is not PASSIVE_NO_RESULT and \
+                child is not NEVER_SET:
             child_state, child_dict = instance_state(child),\
                 instance_dict(child)
             child_impl = child_state.manager[key].impl
diff --git a/lib/sqlalchemy/orm/base.py b/lib/sqlalchemy/orm/base.py
index e06e1fc78..62b4f59a9 100644
--- a/lib/sqlalchemy/orm/base.py
+++ b/lib/sqlalchemy/orm/base.py
@@ -104,6 +104,12 @@ NO_AUTOFLUSH = util.symbol(
     canonical=64
 )
 
+NO_RAISE = util.symbol(
+    "NO_RAISE",
+    """Loader callables should not raise any assertions""",
+    canonical=128
+)
+
 # pre-packaged sets of flags used as inputs
 PASSIVE_OFF = util.symbol(
     "PASSIVE_OFF",
diff --git a/lib/sqlalchemy/orm/strategies.py b/lib/sqlalchemy/orm/strategies.py
index b9abf0647..9d83952d3 100644
--- a/lib/sqlalchemy/orm/strategies.py
+++ b/lib/sqlalchemy/orm/strategies.py
@@ -585,11 +585,14 @@ class LazyLoader(AbstractRelationshipLoader, util.MemoizedSlots):
         ):
             return attributes.PASSIVE_NO_RESULT
 
-        if self._raise_always:
+        if self._raise_always and not passive & attributes.NO_RAISE:
             self._invoke_raise_load(state, passive, "raise")
 
         session = _state_session(state)
         if not session:
+            if passive & attributes.NO_RAISE:
+                return attributes.PASSIVE_NO_RESULT
+
             raise orm_exc.DetachedInstanceError(
                 "Parent instance %s is not bound to a Session; "
                 "lazy load operation of attribute '%s' cannot proceed" %
diff --git a/test/orm/test_relationships.py b/test/orm/test_relationships.py
index ce6d77d91..b60d73c8f 100644
--- a/test/orm/test_relationships.py
+++ b/test/orm/test_relationships.py
@@ -15,6 +15,7 @@ from sqlalchemy.testing import eq_, startswith_, AssertsCompiledSQL, is_, in_
 from sqlalchemy.testing import fixtures
 from test.orm import _fixtures
 from sqlalchemy import exc
+from sqlalchemy.orm import exc as orm_exc
 from sqlalchemy import inspect
 from sqlalchemy import ForeignKeyConstraint
 from sqlalchemy.ext.declarative import declarative_base
@@ -4199,6 +4200,222 @@ class ActiveHistoryFlagTest(_fixtures.FixtureTest):
         self._test_attribute(o1, "composite", MyComposite('bar', 1))
 
 
+class InactiveHistoryNoRaiseTest(_fixtures.FixtureTest):
+    run_inserts = None
+
+    def _run_test(self, detached, raiseload, backref, active_history,
+                  delete):
+
+        if delete:
+            assert not backref, "delete and backref are mutually exclusive"
+
+        Address, addresses, users, User = (self.classes.Address,
+                                           self.tables.addresses,
+                                           self.tables.users,
+                                           self.classes.User)
+
+        opts = {}
+        if active_history:
+            opts['active_history'] = True
+        if raiseload:
+            opts['lazy'] = "raise"
+
+        mapper(Address, addresses, properties={
+            'user': relationship(
+                User, back_populates="addresses", **opts)
+        })
+        mapper(User, users, properties={
+            "addresses": relationship(Address, back_populates="user")
+        })
+
+        s = Session()
+
+        a1 = Address(email_address='a1')
+        u1 = User(name='u1', addresses=[a1])
+        s.add_all([a1, u1])
+        s.commit()
+
+        if backref:
+            u1.addresses
+
+            if detached:
+                s.expunge(a1)
+
+            def go():
+                u1.addresses = []
+
+            if active_history:
+                if raiseload:
+                    assert_raises_message(
+                        exc.InvalidRequestError,
+                        "'Address.user' is not available due to lazy='raise'",
+                        go
+                    )
+                    return
+                elif detached:
+                    assert_raises_message(
+                        orm_exc.DetachedInstanceError,
+                        "lazy load operation of attribute 'user' "
+                        "cannot proceed",
+                        go
+                    )
+                    return
+            go()
+        else:
+            if detached:
+                s.expunge(a1)
+
+            if delete:
+                def go():
+                    del a1.user
+            else:
+                def go():
+                    a1.user = None
+
+            if active_history:
+                if raiseload:
+                    assert_raises_message(
+                        exc.InvalidRequestError,
+                        "'Address.user' is not available due to lazy='raise'",
+                        go
+                    )
+                    return
+                elif detached:
+                    assert_raises_message(
+                        orm_exc.DetachedInstanceError,
+                        "lazy load operation of attribute 'user' "
+                        "cannot proceed",
+                        go
+                    )
+                    return
+            go()
+
+        if detached:
+            s.add(a1)
+
+        s.commit()
+
+    def test_replace_m2o(self):
+        self._run_test(
+            detached=False, raiseload=False,
+            backref=False, delete=False, active_history=False)
+
+    def test_replace_m2o_detached(self):
+        self._run_test(
+            detached=True, raiseload=False,
+            backref=False, delete=False, active_history=False)
+
+    def test_replace_m2o_raiseload(self):
+        self._run_test(
+            detached=False, raiseload=True,
+            backref=False, delete=False, active_history=False)
+
+    def test_replace_m2o_detached_raiseload(self):
+        self._run_test(
+            detached=True, raiseload=True,
+            backref=False, delete=False, active_history=False)
+
+    def test_replace_m2o_backref(self):
+        self._run_test(
+            detached=False, raiseload=False,
+            backref=True, delete=False, active_history=False)
+
+    def test_replace_m2o_detached_backref(self):
+        self._run_test(
+            detached=True, raiseload=False,
+            backref=True, delete=False, active_history=False)
+
+    def test_replace_m2o_raiseload_backref(self):
+        self._run_test(
+            detached=False, raiseload=True,
+            backref=True, delete=False, active_history=False)
+
+    def test_replace_m2o_detached_raiseload_backref(self):
+        self._run_test(
+            detached=True, raiseload=True,
+            backref=True, delete=False, active_history=False)
+
+    def test_replace_m2o_activehistory(self):
+        self._run_test(
+            detached=False, raiseload=False,
+            backref=False, delete=False, active_history=True)
+
+    def test_replace_m2o_detached_activehistory(self):
+        self._run_test(
+            detached=True, raiseload=False,
+            backref=False, delete=False, active_history=True)
+
+    def test_replace_m2o_raiseload_activehistory(self):
+        self._run_test(
+            detached=False, raiseload=True,
+            backref=False, delete=False, active_history=True)
+
+    def test_replace_m2o_detached_raiseload_activehistory(self):
+        self._run_test(
+            detached=True, raiseload=True,
+            backref=False, delete=False, active_history=True)
+
+    def test_replace_m2o_backref_activehistory(self):
+        self._run_test(
+            detached=False, raiseload=False,
+            backref=True, delete=False, active_history=True)
+
+    def test_replace_m2o_detached_backref_activehistory(self):
+        self._run_test(
+            detached=True, raiseload=False,
+            backref=True, delete=False, active_history=True)
+
+    def test_replace_m2o_raiseload_backref_activehistory(self):
+        self._run_test(
+            detached=False, raiseload=True,
+            backref=True, delete=False, active_history=True)
+
+    def test_replace_m2o_detached_raiseload_backref_activehistory(self):
+        self._run_test(
+            detached=True, raiseload=True,
+            backref=True, delete=False, active_history=True)
+
+    def test_delete_m2o(self):
+        self._run_test(
+            detached=False, raiseload=False,
+            backref=False, delete=True, active_history=False)
+
+    def test_delete_m2o_detached(self):
+        self._run_test(
+            detached=True, raiseload=False,
+            backref=False, delete=True, active_history=False)
+
+    def test_delete_m2o_raiseload(self):
+        self._run_test(
+            detached=False, raiseload=True,
+            backref=False, delete=True, active_history=False)
+
+    def test_delete_m2o_detached_raiseload(self):
+        self._run_test(
+            detached=True, raiseload=True,
+            backref=False, delete=True, active_history=False)
+
+    def test_delete_m2o_activehistory(self):
+        self._run_test(
+            detached=False, raiseload=False,
+            backref=False, delete=True, active_history=True)
+
+    def test_delete_m2o_detached_activehistory(self):
+        self._run_test(
+            detached=True, raiseload=False,
+            backref=False, delete=True, active_history=True)
+
+    def test_delete_m2o_raiseload_activehistory(self):
+        self._run_test(
+            detached=False, raiseload=True,
+            backref=False, delete=True, active_history=True)
+
+    def test_delete_m2o_detached_raiseload_activehistory(self):
+        self._run_test(
+            detached=True, raiseload=True,
+            backref=False, delete=True, active_history=True)
+
+
 class RelationDeprecationTest(fixtures.MappedTest):
 
     """test usage of the old 'relation' function."""