Merge "add new pattern for single inh column override" into main

2 files changed, 151 insertions, 50 deletions

diff --git a/test/ext/declarative/test_reflection.py b/test/ext/declarative/test_reflection.py
index e143ad127..53f518a27 100644
--- a/test/ext/declarative/test_reflection.py
+++ b/test/ext/declarative/test_reflection.py
@@ -1,3 +1,6 @@
+from __future__ import annotations
+
+from sqlalchemy import exc
 from sqlalchemy import ForeignKey
 from sqlalchemy import Integer
 from sqlalchemy import String
@@ -7,11 +10,14 @@ from sqlalchemy.orm import clear_mappers
 from sqlalchemy.orm import decl_api as decl
 from sqlalchemy.orm import declared_attr
 from sqlalchemy.orm import exc as orm_exc
+from sqlalchemy.orm import Mapped
+from sqlalchemy.orm import mapped_column
 from sqlalchemy.orm import relationship
 from sqlalchemy.orm import Session
 from sqlalchemy.orm.decl_base import _DeferredMapperConfig
 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.fixtures import fixture_session
 from sqlalchemy.testing.schema import Column
@@ -33,6 +39,10 @@ class DeclarativeReflectionBase(fixtures.TablesTest):
     def teardown_test(self):
         clear_mappers()
 
+    @testing.fixture
+    def decl_base(self):
+        yield Base
+
 
 class DeferredReflectBase(DeclarativeReflectionBase):
     def teardown_test(self):
@@ -346,8 +356,8 @@ class DeferredSingleInhReflectionTest(DeferredInhReflectBase):
             Column("bar_data", String(30)),
         )
 
-    def test_basic(self):
-        class Foo(DeferredReflection, fixtures.ComparableEntity, Base):
+    def test_basic(self, decl_base):
+        class Foo(DeferredReflection, fixtures.ComparableEntity, decl_base):
             __tablename__ = "foo"
             __mapper_args__ = {
                 "polymorphic_on": "type",
@@ -360,8 +370,8 @@ class DeferredSingleInhReflectionTest(DeferredInhReflectBase):
         DeferredReflection.prepare(testing.db)
         self._roundtrip()
 
-    def test_add_subclass_column(self):
-        class Foo(DeferredReflection, fixtures.ComparableEntity, Base):
+    def test_add_subclass_column(self, decl_base):
+        class Foo(DeferredReflection, fixtures.ComparableEntity, decl_base):
             __tablename__ = "foo"
             __mapper_args__ = {
                 "polymorphic_on": "type",
@@ -375,8 +385,40 @@ class DeferredSingleInhReflectionTest(DeferredInhReflectBase):
         DeferredReflection.prepare(testing.db)
         self._roundtrip()
 
-    def test_add_pk_column(self):
-        class Foo(DeferredReflection, fixtures.ComparableEntity, Base):
+    def test_add_subclass_mapped_column(self, decl_base):
+        class Foo(DeferredReflection, fixtures.ComparableEntity, decl_base):
+            __tablename__ = "foo"
+            __mapper_args__ = {
+                "polymorphic_on": "type",
+                "polymorphic_identity": "foo",
+            }
+
+        class Bar(Foo):
+            __mapper_args__ = {"polymorphic_identity": "bar"}
+            bar_data: Mapped[str]
+
+        DeferredReflection.prepare(testing.db)
+        self._roundtrip()
+
+    def test_subclass_mapped_column_no_existing(self, decl_base):
+        class Foo(DeferredReflection, fixtures.ComparableEntity, decl_base):
+            __tablename__ = "foo"
+            __mapper_args__ = {
+                "polymorphic_on": "type",
+                "polymorphic_identity": "foo",
+            }
+
+        with expect_raises_message(
+            exc.ArgumentError,
+            "Can't use use_existing_column with deferred mappers",
+        ):
+
+            class Bar(Foo):
+                __mapper_args__ = {"polymorphic_identity": "bar"}
+                bar_data: Mapped[str] = mapped_column(use_existing_column=True)
+
+    def test_add_pk_column(self, decl_base):
+        class Foo(DeferredReflection, fixtures.ComparableEntity, decl_base):
             __tablename__ = "foo"
             __mapper_args__ = {
                 "polymorphic_on": "type",
@@ -390,6 +432,21 @@ class DeferredSingleInhReflectionTest(DeferredInhReflectBase):
         DeferredReflection.prepare(testing.db)
         self._roundtrip()
 
+    def test_add_pk_mapped_column(self, decl_base):
+        class Foo(DeferredReflection, fixtures.ComparableEntity, decl_base):
+            __tablename__ = "foo"
+            __mapper_args__ = {
+                "polymorphic_on": "type",
+                "polymorphic_identity": "foo",
+            }
+            id: Mapped[int] = mapped_column(primary_key=True)
+
+        class Bar(Foo):
+            __mapper_args__ = {"polymorphic_identity": "bar"}
+
+        DeferredReflection.prepare(testing.db)
+        self._roundtrip()
+
 
 class DeferredJoinedInhReflectionTest(DeferredInhReflectBase):
     @classmethod
diff --git a/test/orm/declarative/test_inheritance.py b/test/orm/declarative/test_inheritance.py
index 9829f4233..f3506a310 100644
--- a/test/orm/declarative/test_inheritance.py
+++ b/test/orm/declarative/test_inheritance.py
@@ -1,5 +1,6 @@
 import sqlalchemy as sa
 from sqlalchemy import ForeignKey
+from sqlalchemy import Identity
 from sqlalchemy import Integer
 from sqlalchemy import select
 from sqlalchemy import String
@@ -9,7 +10,10 @@ from sqlalchemy.orm import close_all_sessions
 from sqlalchemy.orm import configure_mappers
 from sqlalchemy.orm import declared_attr
 from sqlalchemy.orm import deferred
+from sqlalchemy.orm import Mapped
+from sqlalchemy.orm import mapped_column
 from sqlalchemy.orm import relationship
+from sqlalchemy.orm import Session
 from sqlalchemy.orm import with_polymorphic
 from sqlalchemy.orm.decl_api import registry
 from sqlalchemy.testing import assert_raises
@@ -805,11 +809,10 @@ class DeclarativeInheritanceTest(DeclarativeTestBase):
             [Person.__table__.c.name, Person.__table__.c.primary_language],
         )
 
-    @testing.skip_if(
-        lambda: testing.against("oracle"),
-        "Test has an empty insert in it at the moment",
-    )
-    def test_columns_single_inheritance_conflict_resolution(self):
+    @testing.variation("decl_type", ["legacy", "use_existing_column"])
+    def test_columns_single_inheritance_conflict_resolution(
+        self, connection, decl_base, decl_type
+    ):
         """Test that a declared_attr can return the existing column and it will
         be ignored.  this allows conditional columns to be added.
 
@@ -817,18 +820,25 @@ class DeclarativeInheritanceTest(DeclarativeTestBase):
 
         """
 
-        class Person(Base):
+        class Person(decl_base):
             __tablename__ = "person"
-            id = Column(Integer, primary_key=True)
+            id = Column(Integer, Identity(), primary_key=True)
 
         class Engineer(Person):
 
             """single table inheritance"""
 
-            @declared_attr
-            def target_id(cls):
-                return cls.__table__.c.get(
-                    "target_id", Column(Integer, ForeignKey("other.id"))
+            if decl_type.legacy:
+
+                @declared_attr
+                def target_id(cls):
+                    return cls.__table__.c.get(
+                        "target_id", Column(Integer, ForeignKey("other.id"))
+                    )
+
+            elif decl_type.use_existing_column:
+                target_id: Mapped[int] = mapped_column(
+                    ForeignKey("other.id"), use_existing_column=True
                 )
 
             @declared_attr
@@ -839,19 +849,26 @@ class DeclarativeInheritanceTest(DeclarativeTestBase):
 
             """single table inheritance"""
 
-            @declared_attr
-            def target_id(cls):
-                return cls.__table__.c.get(
-                    "target_id", Column(Integer, ForeignKey("other.id"))
+            if decl_type.legacy:
+
+                @declared_attr
+                def target_id(cls):
+                    return cls.__table__.c.get(
+                        "target_id", Column(Integer, ForeignKey("other.id"))
+                    )
+
+            elif decl_type.use_existing_column:
+                target_id: Mapped[int] = mapped_column(
+                    ForeignKey("other.id"), use_existing_column=True
                 )
 
             @declared_attr
             def target(cls):
                 return relationship("Other")
 
-        class Other(Base):
+        class Other(decl_base):
             __tablename__ = "other"
-            id = Column(Integer, primary_key=True)
+            id = Column(Integer, Identity(), primary_key=True)
 
         is_(
             Engineer.target_id.property.columns[0],
@@ -861,22 +878,25 @@ class DeclarativeInheritanceTest(DeclarativeTestBase):
             Manager.target_id.property.columns[0], Person.__table__.c.target_id
         )
         # do a brief round trip on this
-        Base.metadata.create_all(testing.db)
-        session = fixture_session()
-        o1, o2 = Other(), Other()
-        session.add_all(
-            [Engineer(target=o1), Manager(target=o2), Manager(target=o1)]
-        )
-        session.commit()
-        eq_(session.query(Engineer).first().target, o1)
+        decl_base.metadata.create_all(connection)
+        with Session(connection) as session:
+            o1, o2 = Other(), Other()
+            session.add_all(
+                [Engineer(target=o1), Manager(target=o2), Manager(target=o1)]
+            )
+            session.commit()
+            eq_(session.query(Engineer).first().target, o1)
 
-    def test_columns_single_inheritance_conflict_resolution_pk(self):
+    @testing.variation("decl_type", ["legacy", "use_existing_column"])
+    def test_columns_single_inheritance_conflict_resolution_pk(
+        self, decl_base, decl_type
+    ):
         """Test #2472 in terms of a primary key column.  This is
         #4352.
 
         """
 
-        class Person(Base):
+        class Person(decl_base):
             __tablename__ = "person"
             id = Column(Integer, primary_key=True)
 
@@ -886,20 +906,34 @@ class DeclarativeInheritanceTest(DeclarativeTestBase):
 
             """single table inheritance"""
 
-            @declared_attr
-            def target_id(cls):
-                return cls.__table__.c.get(
-                    "target_id", Column(Integer, primary_key=True)
+            if decl_type.legacy:
+
+                @declared_attr
+                def target_id(cls):
+                    return cls.__table__.c.get(
+                        "target_id", Column(Integer, primary_key=True)
+                    )
+
+            elif decl_type.use_existing_column:
+                target_id: Mapped[int] = mapped_column(
+                    primary_key=True, use_existing_column=True
                 )
 
         class Manager(Person):
 
             """single table inheritance"""
 
-            @declared_attr
-            def target_id(cls):
-                return cls.__table__.c.get(
-                    "target_id", Column(Integer, primary_key=True)
+            if decl_type.legacy:
+
+                @declared_attr
+                def target_id(cls):
+                    return cls.__table__.c.get(
+                        "target_id", Column(Integer, primary_key=True)
+                    )
+
+            elif decl_type.use_existing_column:
+                target_id: Mapped[int] = mapped_column(
+                    primary_key=True, use_existing_column=True
                 )
 
         is_(
@@ -910,20 +944,30 @@ class DeclarativeInheritanceTest(DeclarativeTestBase):
             Manager.target_id.property.columns[0], Person.__table__.c.target_id
         )
 
-    def test_columns_single_inheritance_cascading_resolution_pk(self):
+    @testing.variation("decl_type", ["legacy", "use_existing_column"])
+    def test_columns_single_inheritance_cascading_resolution_pk(
+        self, decl_type
+    ):
         """An additional test for #4352 in terms of the requested use case."""
 
         class TestBase(Base):
             __abstract__ = True
 
-            @declared_attr.cascading
-            def id(cls):
-                col_val = None
-                if TestBase not in cls.__bases__:
-                    col_val = cls.__table__.c.get("id")
-                if col_val is None:
-                    col_val = Column(Integer, primary_key=True)
-                return col_val
+            if decl_type.legacy:
+
+                @declared_attr.cascading
+                def id(cls):  # noqa: A001
+                    col_val = None
+                    if TestBase not in cls.__bases__:
+                        col_val = cls.__table__.c.get("id")
+                    if col_val is None:
+                        col_val = Column(Integer, primary_key=True)
+                    return col_val
+
+            elif decl_type.use_existing_column:
+                id: Mapped[int] = mapped_column(  # noqa: A001
+                    primary_key=True, use_existing_column=True
+                )
 
         class Person(TestBase):
             """single table base class"""