Add bulk_replace event, integrate with @validates

Added new attribute event :meth:`.AttributeEvents.bulk_replace`.
This event is triggered when a collection is assigned to a
relationship, before the incoming collection is compared with the
existing one.  This early event allows for conversion of incoming
non-ORM objects as well.  The event is integrated with the
``@validates`` decorator.

The ``@validates`` decorator now allows the decorated method to receive
objects from a "bulk collection set" operation that have not yet
been compared to the existing collection.  This allows incoming values
to be converted to compatible ORM objects as is already allowed
from an "append" event.   Note that this means that the
``@validates`` method is called for **all** values during a collection
assignment, rather than just the ones that are new.

Change-Id: I27f59db008d9e521d31a3e30143d7cd997e4b7b3
Fixes: #3896

4 files changed, 96 insertions, 19 deletions

diff --git a/lib/sqlalchemy/orm/attributes.py b/lib/sqlalchemy/orm/attributes.py
index fc81db782..2b8b38d58 100644
--- a/lib/sqlalchemy/orm/attributes.py
+++ b/lib/sqlalchemy/orm/attributes.py
@@ -324,6 +324,7 @@ def create_proxied_attribute(descriptor):
 OP_REMOVE = util.symbol("REMOVE")
 OP_APPEND = util.symbol("APPEND")
 OP_REPLACE = util.symbol("REPLACE")
+OP_BULK_REPLACE = util.symbol("BULK_REPLACE")
 
 
 class Event(object):
@@ -348,8 +349,9 @@ class Event(object):
     :var impl: The :class:`.AttributeImpl` which is the current event
      initiator.
 
-    :var op: The symbol :attr:`.OP_APPEND`, :attr:`.OP_REMOVE` or
-     :attr:`.OP_REPLACE`, indicating the source operation.
+    :var op: The symbol :attr:`.OP_APPEND`, :attr:`.OP_REMOVE`,
+     :attr:`.OP_REPLACE`, or :attr:`.OP_BULK_REPLACE`, indicating the
+     source operation.
 
     """
 
@@ -1062,6 +1064,10 @@ class CollectionAttributeImpl(AttributeImpl):
                     iterable = iter(iterable)
         new_values = list(iterable)
 
+        evt = Event(self, OP_BULK_REPLACE)
+
+        self.dispatch.bulk_replace(state, new_values, evt)
+
         old = self.get(state, dict_, passive=PASSIVE_ONLY_PERSISTENT)
         if old is PASSIVE_NO_RESULT:
             old = self.initialize(state, dict_)
@@ -1078,7 +1084,8 @@ class CollectionAttributeImpl(AttributeImpl):
         dict_[self.key] = user_data
 
         collections.bulk_replace(
-            new_values, old_collection, new_collection)
+            new_values, old_collection, new_collection,
+            initiator=evt)
 
         del old._sa_adapter
         self.dispatch.dispose_collection(state, old, old_collection)
@@ -1163,7 +1170,7 @@ def backref_listeners(attribute, key, uselist):
             impl = old_state.manager[key].impl
 
             if initiator.impl is not impl or \
-                    initiator.op not in (OP_REPLACE, OP_REMOVE):
+                    initiator.op is OP_APPEND:
                 impl.pop(old_state,
                          old_dict,
                          state.obj(),
@@ -1179,7 +1186,7 @@ def backref_listeners(attribute, key, uselist):
                     initiator.parent_token is not child_impl.parent_token:
                 _acceptable_key_err(state, initiator, child_impl)
             elif initiator.impl is not child_impl or \
-                    initiator.op not in (OP_APPEND, OP_REPLACE):
+                    initiator.op is OP_REMOVE:
                 child_impl.append(
                     child_state,
                     child_dict,
@@ -1200,7 +1207,7 @@ def backref_listeners(attribute, key, uselist):
                 initiator.parent_token is not child_impl.parent_token:
             _acceptable_key_err(state, initiator, child_impl)
         elif initiator.impl is not child_impl or \
-                initiator.op not in (OP_APPEND, OP_REPLACE):
+                initiator.op is OP_REMOVE:
             child_impl.append(
                 child_state,
                 child_dict,
@@ -1215,7 +1222,7 @@ def backref_listeners(attribute, key, uselist):
                 instance_dict(child)
             child_impl = child_state.manager[key].impl
             if initiator.impl is not child_impl or \
-                    initiator.op not in (OP_REMOVE, OP_REPLACE):
+                    initiator.op is OP_APPEND:
                 child_impl.pop(
                     child_state,
                     child_dict,
diff --git a/lib/sqlalchemy/orm/collections.py b/lib/sqlalchemy/orm/collections.py
index d949dc8a1..5a141a90f 100644
--- a/lib/sqlalchemy/orm/collections.py
+++ b/lib/sqlalchemy/orm/collections.py
@@ -728,7 +728,7 @@ class CollectionAdapter(object):
         self.attr = getattr(d['owner_cls'], self._key).impl
 
 
-def bulk_replace(values, existing_adapter, new_adapter):
+def bulk_replace(values, existing_adapter, new_adapter, initiator=None):
     """Load a new collection, firing events based on prior like membership.
 
     Appends instances in ``values`` onto the ``new_adapter``. Events will be
@@ -759,13 +759,13 @@ def bulk_replace(values, existing_adapter, new_adapter):
 
     for member in values or ():
         if member in additions:
-            appender(member)
+            appender(member, _sa_initiator=initiator)
         elif member in constants:
             appender(member, _sa_initiator=False)
 
     if existing_adapter:
         for member in removals:
-            existing_adapter.fire_remove_event(member)
+            existing_adapter.fire_remove_event(member, initiator=initiator)
 
 
 def prepare_instrumentation(factory):
diff --git a/lib/sqlalchemy/orm/events.py b/lib/sqlalchemy/orm/events.py
index ceeb31e9c..1ec898f8c 100644
--- a/lib/sqlalchemy/orm/events.py
+++ b/lib/sqlalchemy/orm/events.py
@@ -1899,6 +1899,10 @@ class AttributeEvents(event.Events):
     def append(self, target, value, initiator):
         """Receive a collection append event.
 
+        The append event is invoked for each element as it is appended
+        to the collection.  This occurs for single-item appends as well
+        as for a "bulk replace" operation.
+
         :param target: the object instance receiving the event.
           If the listener is registered with ``raw=True``, this will
           be the :class:`.InstanceState` object.
@@ -1909,16 +1913,63 @@ class AttributeEvents(event.Events):
         :param initiator: An instance of :class:`.attributes.Event`
           representing the initiation of the event.  May be modified
           from its original value by backref handlers in order to control
-          chained event propagation.
-
-          .. versionchanged:: 0.9.0 the ``initiator`` argument is now
-             passed as a :class:`.attributes.Event` object, and may be
-             modified by backref handlers within a chain of backref-linked
-             events.
-
+          chained event propagation, as well as be inspected for information
+          about the source of the event.
         :return: if the event was registered with ``retval=True``,
          the given value, or a new effective value, should be returned.
 
+        .. seealso::
+
+            :meth:`.AttributeEvents.bulk_replace`
+
+        """
+
+    def bulk_replace(self, target, values, initiator):
+        """Receive a collection 'bulk replace' event.
+
+        This event is invoked for a sequence of values as they are incoming
+        to a bulk collection set operation, which can be
+        modified in place before the values are treated as ORM objects.
+        This is an "early hook" that runs before the bulk replace routine
+        attempts to reconcile which objects are already present in the
+        collection and which are being removed by the net replace operation.
+
+        It is typical that this method be combined with use of the
+        :meth:`.AttributeEvents.append` event.    When using both of these
+        events, note that a bulk replace operation will invoke
+        the :meth:`.AttributeEvents.append` event for all new items,
+        even after :meth:`.AttributeEvents.bulk_replace` has been invoked
+        for the collection as a whole.  In order to determine if an
+        :meth:`.AttributeEvents.append` event is part of a bulk replace,
+        use the symbol :attr:`~.attributes.OP_BULK_REPLACE` to test the
+        incoming initiator::
+
+            from sqlalchemy.orm.attributes import OP_BULK_REPLACE
+
+            @event.listens_for(SomeObject.collection, "bulk_replace")
+            def process_collection(target, values, initiator):
+                values[:] = [_make_value(value) for value in values]
+
+            @event.listens_for(SomeObject.collection, "append", retval=True)
+            def process_collection(target, value, initiator):
+                # make sure bulk_replace didn't already do it
+                if initiator is None or initiator.op is not OP_BULK_REPLACE:
+                    return _make_value(value)
+                else:
+                    return value
+
+
+
+        .. versionadded:: 1.2
+
+        :param target: the object instance receiving the event.
+          If the listener is registered with ``raw=True``, this will
+          be the :class:`.InstanceState` object.
+        :param value: a sequence (e.g. a list) of the values being set.  The
+          handler can modify this list in place.
+        :param initiator: An instance of :class:`.attributes.Event`
+          representing the initiation of the event.
+
         """
 
     def remove(self, target, value, initiator):
diff --git a/lib/sqlalchemy/orm/util.py b/lib/sqlalchemy/orm/util.py
index fc0dba5ee..73b0be99c 100644
--- a/lib/sqlalchemy/orm/util.py
+++ b/lib/sqlalchemy/orm/util.py
@@ -91,11 +91,20 @@ def _validator_events(
 
     if include_removes:
         def append(state, value, initiator):
-            if include_backrefs or not detect_is_backref(state, initiator):
+            if (
+                initiator.op is not attributes.OP_BULK_REPLACE and
+                (include_backrefs or not detect_is_backref(state, initiator))
+            ):
                 return validator(state.obj(), key, value, False)
             else:
                 return value
 
+        def bulk_set(state, values, initiator):
+            if include_backrefs or not detect_is_backref(state, initiator):
+                obj = state.obj()
+                values[:] = [
+                    validator(obj, key, value, False) for value in values]
+
         def set_(state, value, oldvalue, initiator):
             if include_backrefs or not detect_is_backref(state, initiator):
                 return validator(state.obj(), key, value, False)
@@ -108,11 +117,20 @@ def _validator_events(
 
     else:
         def append(state, value, initiator):
-            if include_backrefs or not detect_is_backref(state, initiator):
+            if (
+                initiator.op is not attributes.OP_BULK_REPLACE and
+                (include_backrefs or not detect_is_backref(state, initiator))
+            ):
                 return validator(state.obj(), key, value)
             else:
                 return value
 
+        def bulk_set(state, values, initiator):
+            if include_backrefs or not detect_is_backref(state, initiator):
+                obj = state.obj()
+                values[:] = [
+                    validator(obj, key, value) for value in values]
+
         def set_(state, value, oldvalue, initiator):
             if include_backrefs or not detect_is_backref(state, initiator):
                 return validator(state.obj(), key, value)
@@ -120,6 +138,7 @@ def _validator_events(
                 return value
 
     event.listen(desc, 'append', append, raw=True, retval=True)
+    event.listen(desc, 'bulk_replace', bulk_set, raw=True)
     event.listen(desc, 'set', set_, raw=True, retval=True)
     if include_removes:
         event.listen(desc, "remove", remove, raw=True, retval=True)