From: Mike Bayer Date: Thu, 30 Mar 2023 13:00:49 +0000 (-0400) Subject: skip anno-only mixin columns that are overridden on subclasses X-Git-Tag: rel_2_0_8~3^2 X-Git-Url: http://git.ipfire.org/?a=commitdiff_plain;h=b4c1f29413f678c43d03afdaedd18bd18b23d59a;p=thirdparty%2Fsqlalchemy%2Fsqlalchemy.git 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 index 0000000000..ad3099db4a --- /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 d01aad4395..87166142e8 100644 --- 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 cd6b86e5fd..dcaa18e8b8 100644 --- 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 98c496a81f..9d2f0a540c 100644 --- 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):