Warn for runid changing in load events; add restore_load_context flag

Added a new flag :paramref:`.InstanceEvents.restore_load_context` and
:paramref:`.SessionEvents.restore_load_context` which apply to the
:meth:`.InstanceEvents.load`, :meth:`.InstanceEvents.refresh`, and
:meth:`.SessionEvents.loaded_as_persistent` events, which when set will
restore the "load context" of the object after the event hook has been
called.  This ensures that the object remains within the "loader context"
of the load operation that is already ongoing, rather than the object being
transferred to a new load context due to refresh operations which may have
occurred in the event. A warning is now emitted when this condition occurs,
which recommends use of the flag to resolve this case.  The flag is
"opt-in" so that there is no risk introduced to existing applications.

The change additionally adds support for the ``raw=True`` flag to
session lifecycle events.

Fixes: #5129
Change-Id: I2912f48ac8c5636297d63ed383454930e8e9a6a3

1 files changed, 29 insertions, 1 deletions

diff --git a/lib/sqlalchemy/orm/loading.py b/lib/sqlalchemy/orm/loading.py
index b823de4ab..3f087c904 100644
--- a/lib/sqlalchemy/orm/loading.py
+++ b/lib/sqlalchemy/orm/loading.py
@@ -342,6 +342,18 @@ def _setup_entity_query(
         column_collection.append(pd)
 
 
+def _warn_for_runid_changed(state):
+    util.warn(
+        "Loading context for %s has changed within a load/refresh "
+        "handler, suggesting a row refresh operation took place. If this "
+        "event handler is expected to be "
+        "emitting row refresh operations within an existing load or refresh "
+        "operation, set restore_load_context=True when establishing the "
+        "listener to ensure the context remains unchanged when the event "
+        "handler completes." % (state_str(state),)
+    )
+
+
 def _instance_processor(
     mapper,
     context,
@@ -580,15 +592,28 @@ def _instance_processor(
             )
 
             if isnew:
+                # state.runid should be equal to context.runid / runid
+                # here, however for event checks we are being more conservative
+                # and checking against existing run id
+                # assert state.runid == runid
+
+                existing_runid = state.runid
+
                 if loaded_instance:
                     if load_evt:
                         state.manager.dispatch.load(state, context)
+                        if state.runid != existing_runid:
+                            _warn_for_runid_changed(state)
                     if persistent_evt:
-                        loaded_as_persistent(context.session, state.obj())
+                        loaded_as_persistent(context.session, state)
+                        if state.runid != existing_runid:
+                            _warn_for_runid_changed(state)
                 elif refresh_evt:
                     state.manager.dispatch.refresh(
                         state, context, only_load_props
                     )
+                    if state.runid != runid:
+                        _warn_for_runid_changed(state)
 
                 if populate_existing or state.modified:
                     if refresh_state and only_load_props:
@@ -624,7 +649,10 @@ def _instance_processor(
 
                 if isnew:
                     if refresh_evt:
+                        existing_runid = state.runid
                         state.manager.dispatch.refresh(state, context, to_load)
+                        if state.runid != existing_runid:
+                            _warn_for_runid_changed(state)
 
                     state._commit(dict_, to_load)