- Fixed regression in the :mod:`sqlalchemy.ext.mutable` extension

as a result of the bugfix for :ticket:`3167`,
where attribute and validation events are no longer
called within the flush process.  The mutable
extension was relying upon this behavior in the case where a column
level Python-side default were responsible for generating the new value
on INSERT or UPDATE, or when a value were fetched from the RETURNING
clause for "eager defaults" mode.  The new value would not be subject
to any event when populated and the mutable extension could not
establish proper coercion or history listening.  A new event
:meth:`.InstanceEvents.refresh_flush` is added which the mutable
extension now makes use of for this use case.
fixes #3427
- Added new event :meth:`.InstanceEvents.refresh_flush`, invoked
when an INSERT or UPDATE level default value fetched via RETURNING
or Python-side default is invoked within the flush process.  This
is to provide a hook that is no longer present as a result of
:ticket:`3167`, where attribute and validation events are no longer
called within the flush process.
- Added a new semi-public method to :class:`.MutableBase`
:meth:`.MutableBase._get_listen_keys`.  Overriding this method
is needed in the case where a :class:`.MutableBase` subclass needs
events to propagate for attribute keys other than the key to which
the mutable type is associated with, when intercepting the
:meth:`.InstanceEvents.refresh` or
:meth:`.InstanceEvents.refresh_flush` events.  The current example of
this is composites using :class:`.MutableComposite`.

7 files changed, 286 insertions, 23 deletions

diff --git a/doc/build/changelog/changelog_10.rst b/doc/build/changelog/changelog_10.rst
index f4091a988..291382ba1 100644
--- a/doc/build/changelog/changelog_10.rst
+++ b/doc/build/changelog/changelog_10.rst
@@ -19,6 +19,47 @@
     :version: 1.0.5
 
     .. change::
+        :tags: bug, ext
+        :tickets: 3427
+
+        Fixed regression in the :mod:`sqlalchemy.ext.mutable` extension
+        as a result of the bugfix for :ticket:`3167`,
+        where attribute and validation events are no longer
+        called within the flush process.  The mutable
+        extension was relying upon this behavior in the case where a column
+        level Python-side default were responsible for generating the new value
+        on INSERT or UPDATE, or when a value were fetched from the RETURNING
+        clause for "eager defaults" mode.  The new value would not be subject
+        to any event when populated and the mutable extension could not
+        establish proper coercion or history listening.  A new event
+        :meth:`.InstanceEvents.refresh_flush` is added which the mutable
+        extension now makes use of for this use case.
+
+    .. change::
+        :tags: feature, orm
+        :tickets: 3427
+
+        Added new event :meth:`.InstanceEvents.refresh_flush`, invoked
+        when an INSERT or UPDATE level default value fetched via RETURNING
+        or Python-side default is invoked within the flush process.  This
+        is to provide a hook that is no longer present as a result of
+        :ticket:`3167`, where attribute and validation events are no longer
+        called within the flush process.
+
+    .. change::
+        :tags: feature, ext
+        :tickets: 3427
+
+        Added a new semi-public method to :class:`.MutableBase`
+        :meth:`.MutableBase._get_listen_keys`.  Overriding this method
+        is needed in the case where a :class:`.MutableBase` subclass needs
+        events to propagate for attribute keys other than the key to which
+        the mutable type is associated with, when intercepting the
+        :meth:`.InstanceEvents.refresh` or
+        :meth:`.InstanceEvents.refresh_flush` events.  The current example of
+        this is composites using :class:`.MutableComposite`.
+
+    .. change::
         :tags: bug, engine
         :tickets: 3421
 
diff --git a/lib/sqlalchemy/ext/mutable.py b/lib/sqlalchemy/ext/mutable.py
index 24fc37a42..501b18f39 100644
--- a/lib/sqlalchemy/ext/mutable.py
+++ b/lib/sqlalchemy/ext/mutable.py
@@ -403,6 +403,27 @@ class MutableBase(object):
         raise ValueError(msg % (key, type(value)))
 
     @classmethod
+    def _get_listen_keys(cls, attribute):
+        """Given a descriptor attribute, return a ``set()`` of the attribute
+        keys which indicate a change in the state of this attribute.
+
+        This is normally just ``set([attribute.key])``, but can be overridden
+        to provide for additional keys.  E.g. a :class:`.MutableComposite`
+        augments this set with the attribute keys associated with the columns
+        that comprise the composite value.
+
+        This collection is consulted in the case of intercepting the
+        :meth:`.InstanceEvents.refresh` and
+        :meth:`.InstanceEvents.refresh_flush` events, which pass along a list
+        of attribute names that have been refreshed; the list is compared
+        against this set to determine if action needs to be taken.
+
+        .. versionadded:: 1.0.5
+
+        """
+        return set([attribute.key])
+
+    @classmethod
     def _listen_on_attribute(cls, attribute, coerce, parent_cls):
         """Establish this type as a mutation listener for the given
         mapped descriptor.
@@ -415,6 +436,8 @@ class MutableBase(object):
         # rely on "propagate" here
         parent_cls = attribute.class_
 
+        listen_keys = cls._get_listen_keys(attribute)
+
         def load(state, *args):
             """Listen for objects loaded or refreshed.
 
@@ -429,6 +452,10 @@ class MutableBase(object):
                     state.dict[key] = val
                 val._parents[state.obj()] = key
 
+        def load_attrs(state, ctx, attrs):
+            if not attrs or listen_keys.intersection(attrs):
+                load(state)
+
         def set(target, value, oldvalue, initiator):
             """Listen for set/replace events on the target
             data member.
@@ -463,7 +490,9 @@ class MutableBase(object):
 
         event.listen(parent_cls, 'load', load,
                      raw=True, propagate=True)
-        event.listen(parent_cls, 'refresh', load,
+        event.listen(parent_cls, 'refresh', load_attrs,
+                     raw=True, propagate=True)
+        event.listen(parent_cls, 'refresh_flush', load_attrs,
                      raw=True, propagate=True)
         event.listen(attribute, 'set', set,
                      raw=True, retval=True, propagate=True)
@@ -574,6 +603,10 @@ class MutableComposite(MutableBase):
 
     """
 
+    @classmethod
+    def _get_listen_keys(cls, attribute):
+        return set([attribute.key]).union(attribute.property._attribute_keys)
+
     def changed(self):
         """Subclasses should call this method whenever change events occur."""
 
diff --git a/lib/sqlalchemy/orm/events.py b/lib/sqlalchemy/orm/events.py
index 233cd66a6..801701be9 100644
--- a/lib/sqlalchemy/orm/events.py
+++ b/lib/sqlalchemy/orm/events.py
@@ -272,12 +272,35 @@ class InstanceEvents(event.Events):
          object associated with the instance.
         :param context: the :class:`.QueryContext` corresponding to the
          current :class:`.Query` in progress.
-        :param attrs: iterable collection of attribute names which
+        :param attrs: sequence of attribute names which
          were populated, or None if all column-mapped, non-deferred
          attributes were populated.
 
         """
 
+    def refresh_flush(self, target, flush_context, attrs):
+        """Receive an object instance after one or more attributes have
+        been refreshed within the persistence of the object.
+
+        This event is the same as :meth:`.InstanceEvents.refresh` except
+        it is invoked within the unit of work flush process, and the values
+        here typically come from the process of handling an INSERT or
+        UPDATE, such as via the RETURNING clause or from Python-side default
+        values.
+
+        .. versionadded:: 1.0.5
+
+        :param target: the mapped instance.  If
+         the event is configured with ``raw=True``, this will
+         instead be the :class:`.InstanceState` state-management
+         object associated with the instance.
+        :param flush_context: Internal :class:`.UOWTransaction` object
+         which handles the details of the flush.
+        :param attrs: sequence of attribute names which
+         were populated.
+
+        """
+
     def expire(self, target, attrs):
         """Receive an object instance after its attributes or some subset
         have been expired.
@@ -289,7 +312,7 @@ class InstanceEvents(event.Events):
          the event is configured with ``raw=True``, this will
          instead be the :class:`.InstanceState` state-management
          object associated with the instance.
-        :param attrs: iterable collection of attribute
+        :param attrs: sequence of attribute
          names which were expired, or None if all attributes were
          expired.
 
diff --git a/lib/sqlalchemy/orm/persistence.py b/lib/sqlalchemy/orm/persistence.py
index ab2d54d90..b429aa4c1 100644
--- a/lib/sqlalchemy/orm/persistence.py
+++ b/lib/sqlalchemy/orm/persistence.py
@@ -950,6 +950,10 @@ def _postfetch(mapper, uowtransaction, table,
             mapper.version_id_col in mapper._cols_by_table[table]:
         prefetch_cols = list(prefetch_cols) + [mapper.version_id_col]
 
+    refresh_flush = bool(mapper.class_manager.dispatch.refresh_flush)
+    if refresh_flush:
+        load_evt_attrs = []
+
     if returning_cols:
         row = result.context.returned_defaults
         if row is not None:
@@ -957,10 +961,18 @@ def _postfetch(mapper, uowtransaction, table,
                 if col.primary_key:
                     continue
                 dict_[mapper._columntoproperty[col].key] = row[col]
+                if refresh_flush:
+                    load_evt_attrs.append(mapper._columntoproperty[col].key)
 
     for c in prefetch_cols:
         if c.key in params and c in mapper._columntoproperty:
             dict_[mapper._columntoproperty[c].key] = params[c.key]
+            if refresh_flush:
+                load_evt_attrs.append(mapper._columntoproperty[c].key)
+
+    if refresh_flush and load_evt_attrs:
+        mapper.class_manager.dispatch.refresh_flush(
+            state, uowtransaction, load_evt_attrs)
 
     if postfetch_cols:
         state._expire_attributes(state.dict,
diff --git a/test/ext/test_mutable.py b/test/ext/test_mutable.py
index bb22f0bc2..a6bcdc47f 100644
--- a/test/ext/test_mutable.py
+++ b/test/ext/test_mutable.py
@@ -66,23 +66,25 @@ class MyPoint(Point):
         return value
 
 
-class _MutableDictTestBase(object):
-    run_define_tables = 'each'
-
+class _MutableDictTestFixture(object):
     @classmethod
     def _type_fixture(cls):
         return MutableDict
 
-    def setup_mappers(cls):
-        foo = cls.tables.foo
-
-        mapper(Foo, foo)
-
     def teardown(self):
         # clear out mapper events
         Mapper.dispatch._clear()
         ClassManager.dispatch._clear()
-        super(_MutableDictTestBase, self).teardown()
+        super(_MutableDictTestFixture, self).teardown()
+
+
+class _MutableDictTestBase(_MutableDictTestFixture):
+    run_define_tables = 'each'
+
+    def setup_mappers(cls):
+        foo = cls.tables.foo
+
+        mapper(Foo, foo)
 
     def test_coerce_none(self):
         sess = Session()
@@ -212,6 +214,40 @@ class _MutableDictTestBase(object):
         eq_(f1.non_mutable_data, {'a': 'b'})
 
 
+class MutableColumnDefaultTest(_MutableDictTestFixture, fixtures.MappedTest):
+    @classmethod
+    def define_tables(cls, metadata):
+        MutableDict = cls._type_fixture()
+
+        mutable_pickle = MutableDict.as_mutable(PickleType)
+        Table(
+            'foo', metadata,
+            Column(
+                'id', Integer, primary_key=True,
+                test_needs_autoincrement=True),
+            Column('data', mutable_pickle, default={}),
+        )
+
+    def setup_mappers(cls):
+        foo = cls.tables.foo
+
+        mapper(Foo, foo)
+
+    def test_evt_on_flush_refresh(self):
+        # test for #3427
+
+        sess = Session()
+
+        f1 = Foo()
+        sess.add(f1)
+        sess.flush()
+        assert isinstance(f1.data, self._type_fixture())
+        assert f1 not in sess.dirty
+        f1.data['foo'] = 'bar'
+        assert f1 in sess.dirty
+
+
+
 class MutableWithScalarPickleTest(_MutableDictTestBase, fixtures.MappedTest):
 
     @classmethod
@@ -450,6 +486,43 @@ class _CompositeTestBase(object):
         return Point
 
 
+class MutableCompositeColumnDefaultTest(_CompositeTestBase,
+                                        fixtures.MappedTest):
+    @classmethod
+    def define_tables(cls, metadata):
+        Table(
+            'foo', metadata,
+            Column('id', Integer, primary_key=True,
+                   test_needs_autoincrement=True),
+            Column('x', Integer, default=5),
+            Column('y', Integer, default=9),
+            Column('unrelated_data', String(50))
+        )
+
+    @classmethod
+    def setup_mappers(cls):
+        foo = cls.tables.foo
+
+        cls.Point = cls._type_fixture()
+
+        mapper(Foo, foo, properties={
+            'data': composite(cls.Point, foo.c.x, foo.c.y)
+        })
+
+    def test_evt_on_flush_refresh(self):
+        # this still worked prior to #3427 being fixed in any case
+
+        sess = Session()
+
+        f1 = Foo(data=self.Point(None, None))
+        sess.add(f1)
+        sess.flush()
+        eq_(f1.data, self.Point(5, 9))
+        assert f1 not in sess.dirty
+        f1.data.x = 10
+        assert f1 in sess.dirty
+
+
 class MutableCompositesUnpickleTest(_CompositeTestBase, fixtures.MappedTest):
 
     @classmethod
diff --git a/test/orm/test_events.py b/test/orm/test_events.py
index fc510d585..ae7ba98c1 100644
--- a/test/orm/test_events.py
+++ b/test/orm/test_events.py
@@ -15,7 +15,7 @@ from sqlalchemy.testing import AssertsCompiledSQL
 from sqlalchemy.testing.util import gc_collect
 from test.orm import _fixtures
 from sqlalchemy import event
-from sqlalchemy.testing.mock import Mock, call
+from sqlalchemy.testing.mock import Mock, call, ANY
 
 
 class _RemoveListeners(object):
@@ -32,6 +32,13 @@ class _RemoveListeners(object):
 class MapperEventsTest(_RemoveListeners, _fixtures.FixtureTest):
     run_inserts = None
 
+    @classmethod
+    def define_tables(cls, metadata):
+        super(MapperEventsTest, cls).define_tables(metadata)
+        metadata.tables['users'].append_column(
+            Column('extra', Integer, default=5, onupdate=10)
+        )
+
     def test_instance_event_listen(self):
         """test listen targets for instance events"""
 
@@ -92,6 +99,7 @@ class MapperEventsTest(_RemoveListeners, _fixtures.FixtureTest):
             'init_failure',
             'load',
             'refresh',
+            'refresh_flush',
             'expire',
             'before_insert',
             'after_insert',
@@ -132,10 +140,11 @@ class MapperEventsTest(_RemoveListeners, _fixtures.FixtureTest):
         sess.flush()
         expected = [
             'init', 'before_insert',
+            'refresh_flush',
             'after_insert', 'expire',
             'refresh',
             'load',
-            'before_update', 'after_update', 'before_delete',
+            'before_update', 'refresh_flush', 'after_update', 'before_delete',
             'after_delete']
         eq_(canary, expected)
         eq_(named_canary, expected)
@@ -241,15 +250,17 @@ class MapperEventsTest(_RemoveListeners, _fixtures.FixtureTest):
         sess.flush()
         sess.delete(am)
         sess.flush()
-        eq_(canary1, ['init', 'before_insert', 'after_insert',
+        eq_(canary1, ['init', 'before_insert', 'refresh_flush', 'after_insert',
                       'refresh', 'load',
-                      'before_update', 'after_update', 'before_delete',
+                      'before_update', 'refresh_flush',
+                      'after_update', 'before_delete',
                       'after_delete'])
         eq_(canary2, [])
-        eq_(canary3, ['init', 'before_insert', 'after_insert',
+        eq_(canary3, ['init', 'before_insert', 'refresh_flush', 'after_insert',
                       'refresh',
                       'load',
-                      'before_update', 'after_update', 'before_delete',
+                      'before_update', 'refresh_flush',
+                      'after_update', 'before_delete',
                       'after_delete'])
 
     def test_inheritance_subclass_deferred(self):
@@ -279,14 +290,16 @@ class MapperEventsTest(_RemoveListeners, _fixtures.FixtureTest):
         sess.flush()
         sess.delete(am)
         sess.flush()
-        eq_(canary1, ['init', 'before_insert', 'after_insert',
+        eq_(canary1, ['init', 'before_insert', 'refresh_flush', 'after_insert',
                       'refresh', 'load',
-                      'before_update', 'after_update', 'before_delete',
+                      'before_update', 'refresh_flush',
+                      'after_update', 'before_delete',
                       'after_delete'])
         eq_(canary2, [])
-        eq_(canary3, ['init', 'before_insert', 'after_insert',
+        eq_(canary3, ['init', 'before_insert', 'refresh_flush', 'after_insert',
                       'refresh', 'load',
-                      'before_update', 'after_update', 'before_delete',
+                      'before_update', 'refresh_flush',
+                      'after_update', 'before_delete',
                       'after_delete'])
 
     def test_before_after_only_collection(self):
@@ -2016,3 +2029,65 @@ class QueryEventsTest(
             q.all(),
             [(7, 'jack')]
         )
+
+
+class RefreshFlushInReturningTest(fixtures.MappedTest):
+    """test [ticket:3427].
+
+    this is a rework of the test for [ticket:3167] stated
+    in test_unitofworkv2, which tests that returning doesn't trigger
+    attribute events; the test here is *reversed* so that we test that
+    it *does* trigger the new refresh_flush event.
+
+    """
+
+    __backend__ = True
+
+    @classmethod
+    def define_tables(cls, metadata):
+        Table(
+            'test', metadata,
+            Column('id', Integer, primary_key=True,
+                   test_needs_autoincrement=True),
+            Column('prefetch_val', Integer, default=5),
+            Column('returning_val', Integer, server_default="5")
+        )
+
+    @classmethod
+    def setup_classes(cls):
+        class Thing(cls.Basic):
+            pass
+
+    @classmethod
+    def setup_mappers(cls):
+        Thing = cls.classes.Thing
+
+        mapper(Thing, cls.tables.test, eager_defaults=True)
+
+    def test_no_attr_events_flush(self):
+        Thing = self.classes.Thing
+        mock = Mock()
+        event.listen(Thing, "refresh_flush", mock)
+        t1 = Thing()
+        s = Session()
+        s.add(t1)
+        s.flush()
+
+        if testing.requires.returning.enabled:
+            # ordering is deterministic in this test b.c. the routine
+            # appends the "returning" params before the "prefetch"
+            # ones.  if there were more than one attribute in each category,
+            # then we'd have hash order issues.
+            eq_(
+                mock.mock_calls,
+                [call(t1, ANY, ['returning_val', 'prefetch_val'])]
+            )
+        else:
+            eq_(
+                mock.mock_calls,
+                [call(t1, ANY, ['prefetch_val'])]
+            )
+
+        eq_(t1.id, 1)
+        eq_(t1.prefetch_val, 5)
+        eq_(t1.returning_val, 5)
diff --git a/test/orm/test_unitofworkv2.py b/test/orm/test_unitofworkv2.py
index cef71370d..42b774b10 100644
--- a/test/orm/test_unitofworkv2.py
+++ b/test/orm/test_unitofworkv2.py
@@ -1800,7 +1800,13 @@ class LoadersUsingCommittedTest(UOWTest):
 
 
 class NoAttrEventInFlushTest(fixtures.MappedTest):
-    """test [ticket:3167]"""
+    """test [ticket:3167].
+
+    See also RefreshFlushInReturningTest in test/orm/test_events.py which
+    tests the positive case for the refresh_flush event, added in
+    [ticket:3427].
+
+    """
 
     __backend__ = True