skip anno-only mixin columns that are overridden on subclasses

Fixed issue where an annotation-only :class:`_orm.Mapped` directive could
not be used in a Declarative mixin class, without that attribute attempting
to take effect for single- or joined-inheritance subclasses of mapped
classes that had already mapped that attribute on a superclass, producing
conflicting column errors and/or warnings.

Fixes: #9564
Change-Id: I0f92be2ae98a8c45afce3e06d0a7cc61c19a96f4

diff --git a/doc/build/changelog/unreleased_20/9564.rst b/doc/build/changelog/unreleased_20/9564.rst
new file mode 100644 (file)
index 0000000..ad3099d
--- /dev/null
+++ b/doc/build/changelog/unreleased_20/9564.rst
@@ -0,0 +1,10 @@
+.. change::
+    :tags: bug, orm
+    :tickets: 9564
+
+    Fixed issue where an annotation-only :class:`_orm.Mapped` directive could
+    not be used in a Declarative mixin class, without that attribute attempting
+    to take effect for single- or joined-inheritance subclasses of mapped
+    classes that had already mapped that attribute on a superclass, producing
+    conflicting column errors and/or warnings.
+

diff --git a/lib/sqlalchemy/orm/decl_base.py b/lib/sqlalchemy/orm/decl_base.py
index d01aad439533e6399733fdc1c44d0a9670637863..87166142e82c31d43748dd4ececef63130fceb88 100644 (file)
--- a/lib/sqlalchemy/orm/decl_base.py
+++ b/lib/sqlalchemy/orm/decl_base.py
@@ -616,7 +616,7 @@ class _ClassScanMapperConfig(_MapperConfig):
         if not sa_dataclass_metadata_key:
 
             def attribute_is_overridden(key: str, obj: Any) -> bool:
-                return getattr(cls, key) is not obj
+                return getattr(cls, key, obj) is not obj
 
         else:
 
@@ -973,7 +973,10 @@ class _ClassScanMapperConfig(_MapperConfig):
                         # then test if this name is already handled and
                         # otherwise proceed to generate.
                         if not fixed_table:
-                            assert name in collected_attributes
+                            assert (
+                                name in collected_attributes
+                                or attribute_is_overridden(name, None)
+                            )
                         continue
                     else:
                         # here, the attribute is some other kind of
@@ -1300,11 +1303,22 @@ class _ClassScanMapperConfig(_MapperConfig):
         # copy mixin columns to the mapped class
 
         for name, obj, annotation, is_dataclass in attributes_for_class():
+
             if (
                 not fixed_table
                 and obj is None
                 and _is_mapped_annotation(annotation, cls, originating_class)
             ):
+                # obj is None means this is the annotation only path
+
+                if attribute_is_overridden(name, obj):
+                    # perform same "overridden" check as we do for
+                    # Column/MappedColumn, this is how a mixin col is not
+                    # applied to an inherited subclass that does not have
+                    # the mixin.   the anno-only path added here for
+                    # #9564
+                    continue
+
                 collected_annotation = self._collect_annotation(
                     name, annotation, originating_class, True, obj
                 )

diff --git a/test/orm/declarative/test_tm_future_annotations_sync.py b/test/orm/declarative/test_tm_future_annotations_sync.py
index cd6b86e5fd611cd0db4ddfcc5f815840ea80b791..dcaa18e8b85e4a4a7e43f47f4802fee10b7c708a 100644 (file)
--- a/test/orm/declarative/test_tm_future_annotations_sync.py
+++ b/test/orm/declarative/test_tm_future_annotations_sync.py
@@ -2772,6 +2772,74 @@ class AllYourFavoriteHitsTest(fixtures.TestBase, testing.AssertsCompiledSQL):
             "ON company.company_id = person.company_id",
         )
 
+    @testing.variation("anno_type", ["plain", "typemap", "annotated"])
+    @testing.variation("inh_type", ["single", "joined"])
+    def test_mixin_interp_on_inh(self, decl_base, inh_type, anno_type):
+
+        global anno_col
+
+        if anno_type.typemap:
+            anno_col = Annotated[str, 30]
+
+            decl_base.registry.update_type_annotation_map({anno_col: String})
+
+            class Mixin:
+                foo: Mapped[anno_col]
+
+        elif anno_type.annotated:
+            anno_col = Annotated[str, mapped_column(String)]
+
+            class Mixin:
+                foo: Mapped[anno_col]
+
+        else:
+
+            class Mixin:
+                foo: Mapped[str]
+
+        class Employee(Mixin, decl_base):
+            __tablename__ = "employee"
+
+            id: Mapped[int] = mapped_column(primary_key=True)
+            name: Mapped[str]
+            type: Mapped[str]
+
+            __mapper_args__ = {
+                "polymorphic_on": "type",
+                "polymorphic_identity": "employee",
+            }
+
+        class Manager(Employee):
+            if inh_type.joined:
+                __tablename__ = "manager"
+
+                id: Mapped[int] = mapped_column(  # noqa: A001
+                    ForeignKey("employee.id"), primary_key=True
+                )
+
+            manager_data: Mapped[str] = mapped_column(nullable=True)
+
+            __mapper_args__ = {
+                "polymorphic_identity": "manager",
+            }
+
+        if inh_type.single:
+            self.assert_compile(
+                select(Manager),
+                "SELECT employee.id, employee.name, employee.type, "
+                "employee.foo, employee.manager_data FROM employee "
+                "WHERE employee.type IN (__[POSTCOMPILE_type_1])",
+            )
+        elif inh_type.joined:
+            self.assert_compile(
+                select(Manager),
+                "SELECT manager.id, employee.id AS id_1, employee.name, "
+                "employee.type, employee.foo, manager.manager_data "
+                "FROM employee JOIN manager ON employee.id = manager.id",
+            )
+        else:
+            inh_type.fail()
+
 
 class WriteOnlyRelationshipTest(fixtures.TestBase):
     def _assertions(self, A, B, lazy):

diff --git a/test/orm/declarative/test_typed_mapping.py b/test/orm/declarative/test_typed_mapping.py
index 98c496a81f979fd3134769f73d59440e63c85639..9d2f0a540c0f0afcd52f5f655a95aa99ce70a967 100644 (file)
--- a/test/orm/declarative/test_typed_mapping.py
+++ b/test/orm/declarative/test_typed_mapping.py
@@ -2763,6 +2763,74 @@ class AllYourFavoriteHitsTest(fixtures.TestBase, testing.AssertsCompiledSQL):
             "ON company.company_id = person.company_id",
         )
 
+    @testing.variation("anno_type", ["plain", "typemap", "annotated"])
+    @testing.variation("inh_type", ["single", "joined"])
+    def test_mixin_interp_on_inh(self, decl_base, inh_type, anno_type):
+
+        # anno only: global anno_col
+
+        if anno_type.typemap:
+            anno_col = Annotated[str, 30]
+
+            decl_base.registry.update_type_annotation_map({anno_col: String})
+
+            class Mixin:
+                foo: Mapped[anno_col]
+
+        elif anno_type.annotated:
+            anno_col = Annotated[str, mapped_column(String)]
+
+            class Mixin:
+                foo: Mapped[anno_col]
+
+        else:
+
+            class Mixin:
+                foo: Mapped[str]
+
+        class Employee(Mixin, decl_base):
+            __tablename__ = "employee"
+
+            id: Mapped[int] = mapped_column(primary_key=True)
+            name: Mapped[str]
+            type: Mapped[str]
+
+            __mapper_args__ = {
+                "polymorphic_on": "type",
+                "polymorphic_identity": "employee",
+            }
+
+        class Manager(Employee):
+            if inh_type.joined:
+                __tablename__ = "manager"
+
+                id: Mapped[int] = mapped_column(  # noqa: A001
+                    ForeignKey("employee.id"), primary_key=True
+                )
+
+            manager_data: Mapped[str] = mapped_column(nullable=True)
+
+            __mapper_args__ = {
+                "polymorphic_identity": "manager",
+            }
+
+        if inh_type.single:
+            self.assert_compile(
+                select(Manager),
+                "SELECT employee.id, employee.name, employee.type, "
+                "employee.foo, employee.manager_data FROM employee "
+                "WHERE employee.type IN (__[POSTCOMPILE_type_1])",
+            )
+        elif inh_type.joined:
+            self.assert_compile(
+                select(Manager),
+                "SELECT manager.id, employee.id AS id_1, employee.name, "
+                "employee.type, employee.foo, manager.manager_data "
+                "FROM employee JOIN manager ON employee.id = manager.id",
+            )
+        else:
+            inh_type.fail()
+
 
 class WriteOnlyRelationshipTest(fixtures.TestBase):
     def _assertions(self, A, B, lazy):