]> git.ipfire.org Git - thirdparty/sqlalchemy/sqlalchemy.git/commitdiff
Improve check for overlapping FK targets on sibling classes
authorMike Bayer <mike_mp@zzzcomputing.com>
Tue, 3 Oct 2017 00:50:56 +0000 (20:50 -0400)
committerMike Bayer <mike_mp@zzzcomputing.com>
Tue, 3 Oct 2017 18:42:49 +0000 (14:42 -0400)
Fixed bug where ORM relationship would warn against conflicting sync
targets (e.g. two relationships would both write to the same column) for
sibling classes in an inheritance hierarchy, where the two relationships
would never actually conflict during writes.

Change-Id: I9367a7978cadc59066e89fc4917d7eb6c78dedee
Fixes: #4078
(cherry picked from commit 8ba8dd23b7cbf9aa423b6aa965abc4d7174b84de)

doc/build/changelog/unreleased_11/4078.rst [new file with mode: 0644]
lib/sqlalchemy/orm/relationships.py
test/orm/test_relationships.py

diff --git a/doc/build/changelog/unreleased_11/4078.rst b/doc/build/changelog/unreleased_11/4078.rst
new file mode 100644 (file)
index 0000000..0c578b8
--- /dev/null
@@ -0,0 +1,9 @@
+.. change::
+    :tags: bug, orm
+    :tickets: 4078
+    :versions: 1.2.0b3
+
+    Fixed bug where ORM relationship would warn against conflicting sync
+    targets (e.g. two relationships would both write to the same column) for
+    sibling classes in an inheritance hierarchy, where the two relationships
+    would never actually conflict during writes.
\ No newline at end of file
index c46071cb60dd80a15cebb64d19599027919518a8..d86e4544063fabb3b2e8511ef32bda3cdc85f907 100644 (file)
@@ -1799,6 +1799,15 @@ class RelationshipProperty(StrategizedProperty):
                 (self.key, self.parent.class_)
             )
 
+    def _persists_for(self, mapper):
+        """Return True if this property will persist values on behalf
+        of the given mapper.
+
+        """
+
+        return self.key in mapper.relationships and \
+            mapper.relationships[self.key] is self
+
     def _columns_are_mapped(self, *cols):
         """Return True if all columns in the given collection are
         mapped by the tables referenced by this :class:`.Relationship`.
@@ -2673,10 +2682,16 @@ class JoinCondition(object):
             else:
                 other_props = []
                 prop_to_from = self._track_overlapping_sync_targets[to_]
+
                 for pr, fr_ in prop_to_from.items():
                     if pr.mapper in mapperlib._mapper_registry and \
+                        (
+                            self.prop._persists_for(pr.parent) or
+                            pr._persists_for(self.prop.parent)
+                        ) and \
                         fr_ is not from_ and \
                             pr not in self.prop._reverse_property:
+
                         other_props.append((pr, fr_))
 
                 if other_props:
index d2af2b1a7670773a615f0ac711a51f5bae7eac28..ca0cf0a26ad8dff1f3ac4defc2554892b6357a15 100644 (file)
@@ -16,6 +16,8 @@ from sqlalchemy.testing import fixtures
 from test.orm import _fixtures
 from sqlalchemy import exc
 from sqlalchemy import inspect
+from sqlalchemy import ForeignKeyConstraint
+from sqlalchemy.ext.declarative import declarative_base
 
 
 class _RelationshipErrors(object):
@@ -528,6 +530,166 @@ class DirectSelfRefFKTest(fixtures.MappedTest, AssertsCompiledSQL):
         )
 
 
+class OverlappingFksSiblingTest(fixtures.TestBase):
+    """Test multiple relationships that use sections of the same
+    composite foreign key.
+
+    """
+
+    def teardown(self):
+        clear_mappers()
+
+    def _fixture_one(
+            self, add_b_a=False, add_b_a_viewonly=False, add_b_amember=False,
+            add_bsub1_a=False, add_bsub2_a_viewonly=False):
+
+        Base = declarative_base(metadata=self.metadata)
+
+        class A(Base):
+
+            __tablename__ = 'a'
+
+            id = Column(Integer, primary_key=True)
+            a_members = relationship('AMember', backref='a')
+
+        class AMember(Base):
+
+            __tablename__ = 'a_member'
+
+            a_id = Column(Integer, ForeignKey('a.id'), primary_key=True)
+            a_member_id = Column(Integer, primary_key=True)
+
+        class B(Base):
+            __tablename__ = 'b'
+
+            __mapper_args__ = {
+                'polymorphic_on': 'type'
+            }
+
+            id = Column(Integer, primary_key=True)
+            type = Column(String(20))
+
+            a_id = Column(Integer, ForeignKey('a.id'), nullable=False)
+            a_member_id = Column(Integer)
+
+            __table_args__ = (
+                ForeignKeyConstraint(
+                    ('a_id', 'a_member_id'),
+                    ('a_member.a_id', 'a_member.a_member_id')),
+            )
+
+            # if added and viewonly is not true, this relationship
+            # writes to B.a_id, which conflicts with BSub2.a_member,
+            # so should warn
+            if add_b_a:
+                a = relationship('A', viewonly=add_b_a_viewonly)
+
+            # if added, this relationship writes to B.a_id, which conflicts
+            # with BSub1.a
+            if add_b_amember:
+                a_member = relationship('AMember')
+
+        # however, *no* warning should be emitted otherwise.
+
+        class BSub1(B):
+
+            if add_bsub1_a:
+                a = relationship('A')
+
+            __mapper_args__ = {'polymorphic_identity': 'bsub1'}
+
+        class BSub2(B):
+
+            if add_bsub2_a_viewonly:
+                a = relationship("A", viewonly=True)
+
+            a_member = relationship('AMember')
+
+            __mapper_args__ = {'polymorphic_identity': 'bsub2'}
+
+        configure_mappers()
+        self.metadata.create_all()
+
+        return A, AMember, B, BSub1, BSub2
+
+    @testing.provide_metadata
+    def _test_fixture_one_run(self, **kw):
+        A, AMember, B, BSub1, BSub2 = self._fixture_one(**kw)
+
+        bsub2 = BSub2()
+        am1 = AMember(a_member_id=1)
+
+        a1 = A(a_members=[am1])
+        bsub2.a_member = am1
+
+        bsub1 = BSub1()
+        a2 = A()
+        bsub1.a = a2
+
+        session = Session(testing.db)
+        session.add_all([bsub1, bsub2, am1, a1, a2])
+        session.commit()
+
+        assert bsub1.a is a2
+        assert bsub2.a is a1
+
+        # meaningless, because BSub1 doesn't have a_member
+        bsub1.a_member = am1
+
+        # meaningless, because BSub2's ".a" is viewonly=True
+        bsub2.a = a2
+
+        session.commit()
+        assert bsub1.a is a2  # beacuse bsub1.a_member is not a relationship
+        assert bsub2.a is a1  # because bsub2.a is viewonly=True
+
+        # everyone has a B.a relationship
+        eq_(
+            session.query(B, A).outerjoin(B.a).order_by(B.id).all(),
+            [(bsub1, a2), (bsub2, a1)]
+        )
+
+    @testing.provide_metadata
+    def test_warn_one(self):
+        assert_raises_message(
+            exc.SAWarning,
+            r"relationship '(?:BSub1.a|BSub2.a_member|B.a)' will copy column "
+            r"(?:a.id|a_member.a_id) to column b.a_id",
+            self._fixture_one, add_b_a=True, add_bsub1_a=True
+        )
+
+    @testing.provide_metadata
+    def test_warn_two(self):
+        assert_raises_message(
+            exc.SAWarning,
+            r"relationship '(?:BSub1.a|B.a_member)' will copy column "
+            r"(?:a.id|a_member.a_id) to column b.a_id",
+            self._fixture_one, add_b_amember=True, add_bsub1_a=True
+        )
+
+    @testing.provide_metadata
+    def test_warn_three(self):
+        assert_raises_message(
+            exc.SAWarning,
+            r"relationship '(?:BSub1.a|B.a_member|B.a)' will copy column "
+            r"(?:a.id|a_member.a_id) to column b.a_id",
+            self._fixture_one, add_b_amember=True, add_bsub1_a=True,
+            add_b_a=True
+        )
+
+    @testing.provide_metadata
+    def test_works_one(self):
+        self._test_fixture_one_run(
+            add_b_a=True, add_b_a_viewonly=True, add_bsub1_a=True
+        )
+
+    @testing.provide_metadata
+    def test_works_two(self):
+        self._test_fixture_one_run(
+            add_b_a=True, add_bsub2_a_viewonly=True
+        )
+
+
 class CompositeSelfRefFKTest(fixtures.MappedTest, AssertsCompiledSQL):
 
     """Tests a composite FK where, in