Merge "implement active_history for composites" into main

3 files changed, 223 insertions, 2 deletions

diff --git a/doc/build/changelog/unreleased_20/9460.rst b/doc/build/changelog/unreleased_20/9460.rst
new file mode 100644
index 000000000..a2a7d9c6d
--- /dev/null
+++ b/doc/build/changelog/unreleased_20/9460.rst
@@ -0,0 +1,17 @@
+.. change::
+    :tags: bug, orm
+    :tickets: 9460
+
+    Fixed bug where the "active history" feature was not fully
+    implemented for composite attributes, making it impossible to receive
+    events that included the "old" value.   This seems to have been the case
+    with older SQLAlchemy versions as well, where "active_history" would
+    be propagated to the underlying column-based attributes, but an event
+    handler listening to the composite attribute itself would not be given
+    the "old" value being replaced, even if the composite() were set up
+    with active_history=True.
+
+    Additionally, fixed a regression that's local to 2.0 which disallowed
+    active_history on composite from being assigned to the impl with
+    ``attr.impl.active_history=True``.
+
diff --git a/lib/sqlalchemy/orm/descriptor_props.py b/lib/sqlalchemy/orm/descriptor_props.py
index fd28830d9..2e57fd0f4 100644
--- a/lib/sqlalchemy/orm/descriptor_props.py
+++ b/lib/sqlalchemy/orm/descriptor_props.py
@@ -305,7 +305,12 @@ class CompositeProperty(
             dict_ = attributes.instance_dict(instance)
             state = attributes.instance_state(instance)
             attr = state.manager[self.key]
-            previous = dict_.get(self.key, LoaderCallableStatus.NO_VALUE)
+
+            if attr.dispatch._active_history:
+                previous = fget(instance)
+            else:
+                previous = dict_.get(self.key, LoaderCallableStatus.NO_VALUE)
+
             for fn in attr.dispatch.set:
                 value = fn(state, value, previous, attr.impl)
             dict_[self.key] = value
@@ -322,7 +327,14 @@ class CompositeProperty(
         def fdel(instance: Any) -> None:
             state = attributes.instance_state(instance)
             dict_ = attributes.instance_dict(instance)
-            previous = dict_.pop(self.key, LoaderCallableStatus.NO_VALUE)
+            attr = state.manager[self.key]
+
+            if attr.dispatch._active_history:
+                previous = fget(instance)
+                dict_.pop(self.key, None)
+            else:
+                previous = dict_.pop(self.key, LoaderCallableStatus.NO_VALUE)
+
             attr = state.manager[self.key]
             attr.dispatch.remove(state, previous, attr.impl)
             for key in self._attribute_keys:
@@ -614,6 +626,10 @@ class CompositeProperty(
             self.parent, "expire", expire_handler, raw=True, propagate=True
         )
 
+        proxy_attr = self.parent.class_manager[self.key]
+        proxy_attr.impl.dispatch = proxy_attr.dispatch  # type: ignore
+        proxy_attr.impl.dispatch._active_history = self.active_history  # type: ignore  # noqa: E501
+
         # TODO: need a deserialize hook here
 
     @util.memoized_property
diff --git a/test/orm/test_composites.py b/test/orm/test_composites.py
index ddd9a45b8..ded2c25db 100644
--- a/test/orm/test_composites.py
+++ b/test/orm/test_composites.py
@@ -3,8 +3,10 @@ import operator
 import random
 
 import sqlalchemy as sa
+from sqlalchemy import event
 from sqlalchemy import ForeignKey
 from sqlalchemy import insert
+from sqlalchemy import inspect
 from sqlalchemy import Integer
 from sqlalchemy import select
 from sqlalchemy import String
@@ -14,13 +16,16 @@ from sqlalchemy.orm import aliased
 from sqlalchemy.orm import Composite
 from sqlalchemy.orm import composite
 from sqlalchemy.orm import configure_mappers
+from sqlalchemy.orm import mapped_column
 from sqlalchemy.orm import relationship
 from sqlalchemy.orm import Session
+from sqlalchemy.orm.attributes import LoaderCallableStatus
 from sqlalchemy.testing import assert_raises_message
 from sqlalchemy.testing import eq_
 from sqlalchemy.testing import expect_raises_message
 from sqlalchemy.testing import fixtures
 from sqlalchemy.testing import is_
+from sqlalchemy.testing import mock
 from sqlalchemy.testing.fixtures import fixture_session
 from sqlalchemy.testing.schema import Column
 from sqlalchemy.testing.schema import Table
@@ -851,6 +856,189 @@ class NestedTest(fixtures.MappedTest, testing.AssertsCompiledSQL):
         eq_(t1.ab, AB("a", "b", CD("c", "d")))
 
 
+class EventsEtcTest(fixtures.MappedTest):
+    @testing.fixture
+    def point_fixture(self, decl_base):
+        def go(active_history):
+            @dataclasses.dataclass
+            class Point:
+                x: int
+                y: int
+
+            class Edge(decl_base):
+                __tablename__ = "edge"
+                id = mapped_column(Integer, primary_key=True)
+
+                start = composite(
+                    Point,
+                    mapped_column("x1", Integer),
+                    mapped_column("y1", Integer),
+                    active_history=active_history,
+                )
+                end = composite(
+                    Point,
+                    mapped_column("x2", Integer, nullable=True),
+                    mapped_column("y2", Integer, nullable=True),
+                    active_history=active_history,
+                )
+
+            decl_base.metadata.create_all(testing.db)
+
+            return Point, Edge
+
+        return go
+
+    @testing.variation("active_history", [True, False])
+    @testing.variation("hist_on_mapping", [True, False])
+    def test_event_listener_no_value_to_set(
+        self, point_fixture, active_history, hist_on_mapping
+    ):
+        if hist_on_mapping:
+            config_active_history = bool(active_history)
+        else:
+            config_active_history = False
+
+        Point, Edge = point_fixture(config_active_history)
+
+        if not hist_on_mapping and active_history:
+            Edge.start.impl.active_history = True
+
+        m1 = mock.Mock()
+
+        event.listen(Edge.start, "set", m1)
+
+        e1 = Edge()
+        e1.start = Point(5, 6)
+
+        eq_(
+            m1.mock_calls,
+            [
+                mock.call(
+                    e1,
+                    Point(5, 6),
+                    LoaderCallableStatus.NO_VALUE
+                    if not active_history
+                    else None,
+                    Edge.start.impl,
+                )
+            ],
+        )
+
+        eq_(
+            inspect(e1).attrs.start.history,
+            ([Point(5, 6)], (), [Point(None, None)]),
+        )
+
+    @testing.variation("active_history", [True, False])
+    @testing.variation("hist_on_mapping", [True, False])
+    def test_event_listener_set_to_new(
+        self, point_fixture, active_history, hist_on_mapping
+    ):
+        if hist_on_mapping:
+            config_active_history = bool(active_history)
+        else:
+            config_active_history = False
+
+        Point, Edge = point_fixture(config_active_history)
+
+        if not hist_on_mapping and active_history:
+            Edge.start.impl.active_history = True
+
+        e1 = Edge()
+        e1.start = Point(5, 6)
+
+        sess = fixture_session()
+
+        sess.add(e1)
+        sess.commit()
+        assert "start" not in e1.__dict__
+
+        m1 = mock.Mock()
+
+        event.listen(Edge.start, "set", m1)
+
+        e1.start = Point(7, 8)
+
+        eq_(
+            m1.mock_calls,
+            [
+                mock.call(
+                    e1,
+                    Point(7, 8),
+                    LoaderCallableStatus.NO_VALUE
+                    if not active_history
+                    else Point(5, 6),
+                    Edge.start.impl,
+                )
+            ],
+        )
+
+        if active_history:
+            eq_(
+                inspect(e1).attrs.start.history,
+                ([Point(7, 8)], (), [Point(5, 6)]),
+            )
+        else:
+            eq_(
+                inspect(e1).attrs.start.history,
+                ([Point(7, 8)], (), [Point(None, None)]),
+            )
+
+    @testing.variation("active_history", [True, False])
+    @testing.variation("hist_on_mapping", [True, False])
+    def test_event_listener_set_to_deleted(
+        self, point_fixture, active_history, hist_on_mapping
+    ):
+        if hist_on_mapping:
+            config_active_history = bool(active_history)
+        else:
+            config_active_history = False
+
+        Point, Edge = point_fixture(config_active_history)
+
+        if not hist_on_mapping and active_history:
+            Edge.start.impl.active_history = True
+
+        e1 = Edge()
+        e1.start = Point(5, 6)
+
+        sess = fixture_session()
+
+        sess.add(e1)
+        sess.commit()
+        assert "start" not in e1.__dict__
+
+        m1 = mock.Mock()
+
+        event.listen(Edge.start, "remove", m1)
+
+        del e1.start
+
+        eq_(
+            m1.mock_calls,
+            [
+                mock.call(
+                    e1,
+                    LoaderCallableStatus.NO_VALUE
+                    if not active_history
+                    else Point(5, 6),
+                    Edge.start.impl,
+                )
+            ],
+        )
+
+        if active_history:
+            eq_(
+                inspect(e1).attrs.start.history,
+                ([Point(None, None)], (), [Point(5, 6)]),
+            )
+        else:
+            eq_(
+                inspect(e1).attrs.start.history,
+                ([Point(None, None)], (), [Point(None, None)]),
+            )
+
+
 class PrimaryKeyTest(fixtures.MappedTest):
     @classmethod
     def define_tables(cls, metadata):