guard against duplicate mutable event listeners

Fixed issue in :class:`_mutable.Mutable` where event registration for ORM
mapped attributes would be called repeatedly for mapped inheritance
subclasses, leading to duplicate events being invoked in inheritance
hierarchies.

Fixes: #9676
Change-Id: I91289141d7a5f5c86a9033596735ed6eba7071b0

3 files changed, 62 insertions, 18 deletions

diff --git a/doc/build/changelog/unreleased_20/9676.rst b/doc/build/changelog/unreleased_20/9676.rst
new file mode 100644
index 000000000..21edcaab6
--- /dev/null
+++ b/doc/build/changelog/unreleased_20/9676.rst
@@ -0,0 +1,8 @@
+.. change::
+    :tags: bug, ext
+    :tickets: 9676
+
+    Fixed issue in :class:`_mutable.Mutable` where event registration for ORM
+    mapped attributes would be called repeatedly for mapped inheritance
+    subclasses, leading to duplicate events being invoked in inheritance
+    hierarchies.
diff --git a/lib/sqlalchemy/ext/mutable.py b/lib/sqlalchemy/ext/mutable.py
index 7d23f9fda..0f82518aa 100644
--- a/lib/sqlalchemy/ext/mutable.py
+++ b/lib/sqlalchemy/ext/mutable.py
@@ -693,14 +693,28 @@ class Mutable(MutableBase):
         ) -> None:
             if mapper.non_primary:
                 return
+            _APPLIED_KEY = "_ext_mutable_listener_applied"
+
             for prop in mapper.column_attrs:
                 if (
-                    schema_event_check
-                    and hasattr(prop.expression, "info")
-                    and prop.expression.info.get("_ext_mutable_orig_type")  # type: ignore # noqa: E501 # TODO: https://github.com/python/mypy/issues/1424#issuecomment-1272354487
-                    is sqltype
-                ) or (prop.columns[0].type is sqltype):
-                    cls.associate_with_attribute(getattr(class_, prop.key))
+                    # all Mutable types refer to a Column that's mapped,
+                    # since this is the only kind of Core target the ORM can
+                    # "mutate"
+                    isinstance(prop.expression, Column)
+                    and (
+                        (
+                            schema_event_check
+                            and prop.expression.info.get(
+                                "_ext_mutable_orig_type"
+                            )
+                            is sqltype
+                        )
+                        or prop.expression.type is sqltype
+                    )
+                ):
+                    if not prop.expression.info.get(_APPLIED_KEY, False):
+                        prop.expression.info[_APPLIED_KEY] = True
+                        cls.associate_with_attribute(getattr(class_, prop.key))
 
         event.listen(Mapper, "mapper_configured", listen_for_type)
 
@@ -724,7 +738,6 @@ class MutableComposite(MutableBase):
         """Subclasses should call this method whenever change events occur."""
 
         for parent, key in self._parents.items():
-
             prop = parent.mapper.get_property(key)
             for value, attr_name in zip(
                 prop._composite_values_from_instance(self),
@@ -781,7 +794,6 @@ class MutableDict(Mutable, Dict[_KT, _VT]):
         self.changed()
 
     if TYPE_CHECKING:
-
         # from https://github.com/python/mypy/issues/14858
 
         @overload
diff --git a/test/ext/test_mutable.py b/test/ext/test_mutable.py
index 290518dd6..6c428fa85 100644
--- a/test/ext/test_mutable.py
+++ b/test/ext/test_mutable.py
@@ -1,6 +1,10 @@
+from __future__ import annotations
+
 import copy
 import dataclasses
 import pickle
+from typing import Any
+from typing import Dict
 
 from sqlalchemy import event
 from sqlalchemy import ForeignKey
@@ -19,6 +23,8 @@ from sqlalchemy.orm import attributes
 from sqlalchemy.orm import column_property
 from sqlalchemy.orm import composite
 from sqlalchemy.orm import declarative_base
+from sqlalchemy.orm import Mapped
+from sqlalchemy.orm import mapped_column
 from sqlalchemy.orm import Session
 from sqlalchemy.orm.instrumentation import ClassManager
 from sqlalchemy.orm.mapper import Mapper
@@ -168,7 +174,6 @@ class MiscTest(fixtures.TestBase):
         registry.metadata.create_all(connection)
 
         with Session(connection) as sess:
-
             data = dict(
                 j1={"a": 1},
                 j2={"b": 2},
@@ -243,6 +248,34 @@ class MiscTest(fixtures.TestBase):
 
             is_true(inspect(t1_merged).attrs.data.history.added)
 
+    def test_no_duplicate_reg_w_inheritance(self, decl_base):
+        """test #9676"""
+
+        class A(decl_base):
+            __tablename__ = "a"
+
+            id: Mapped[int] = mapped_column(primary_key=True)
+
+            json: Mapped[Dict[str, Any]] = mapped_column(
+                MutableDict.as_mutable(JSON())
+            )
+
+        class B(A):
+            pass
+
+        class C(B):
+            pass
+
+        decl_base.registry.configure()
+
+        # the event hook itself doesnt do anything for repeated calls
+        # already, so there's really nothing else to assert other than there's
+        # only one "set" event listener
+
+        eq_(len(A.json.dispatch.set), 1)
+        eq_(len(B.json.dispatch.set), 1)
+        eq_(len(C.json.dispatch.set), 1)
+
 
 class _MutableDictTestBase(_MutableDictTestFixture):
     run_define_tables = "each"
@@ -1252,7 +1285,6 @@ class MutableAssocWithAttrInheritTest(
 ):
     @classmethod
     def define_tables(cls, metadata):
-
         Table(
             "foo",
             metadata,
@@ -1360,7 +1392,6 @@ class MutableAssociationScalarJSONTest(
 class CustomMutableAssociationScalarJSONTest(
     _MutableDictTestBase, fixtures.MappedTest
 ):
-
     CustomMutableDict = None
 
     @classmethod
@@ -1445,7 +1476,6 @@ class _CompositeTestBase:
 
     @classmethod
     def _type_fixture(cls):
-
         return Point
 
 
@@ -1494,7 +1524,6 @@ class MutableCompositeColumnDefaultTest(
 class MutableDCCompositeColumnDefaultTest(MutableCompositeColumnDefaultTest):
     @classmethod
     def _type_fixture(cls):
-
         return DCPoint
 
 
@@ -1520,7 +1549,6 @@ class MutableCompositesUnpickleTest(_CompositeTestBase, fixtures.MappedTest):
 class MutableDCCompositesUnpickleTest(MutableCompositesUnpickleTest):
     @classmethod
     def _type_fixture(cls):
-
         return DCPoint
 
 
@@ -1638,7 +1666,6 @@ class MutableCompositesTest(_CompositeTestBase, fixtures.MappedTest):
 class MutableDCCompositesTest(MutableCompositesTest):
     @classmethod
     def _type_fixture(cls):
-
         return DCPoint
 
 
@@ -1676,7 +1703,6 @@ class MutableCompositeCustomCoerceTest(
 ):
     @classmethod
     def _type_fixture(cls):
-
         return MyPoint
 
     @classmethod
@@ -1710,7 +1736,6 @@ class MutableCompositeCustomCoerceTest(
 class MutableDCCompositeCustomCoerceTest(MutableCompositeCustomCoerceTest):
     @classmethod
     def _type_fixture(cls):
-
         return MyDCPoint
 
 
@@ -1780,5 +1805,4 @@ class MutableInheritedCompositesTest(_CompositeTestBase, fixtures.MappedTest):
 class MutableInheritedDCCompositesTest(MutableInheritedCompositesTest):
     @classmethod
     def _type_fixture(cls):
-
         return DCPoint