column overlap warning in relationship needs tuning for sibling classes

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.engine import create_engine
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy.schema import Column, ForeignKey, ForeignKeyConstraint
from sqlalchemy.types import Integer, String
Base = declarative_base()


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)

    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 viewonly is removed, warning should emit:
    # "relationship 'B.a' will copy column a.id to column b.a_id, which
    # "conflicts with relationship(s): 'BSub2.a_member' (copies a_member.a_id to b.a_id). ""
    a = relationship('A', viewonly=True)

    # if uncommented, warning should emit:
    # relationship 'B.a_member' will copy column a_member.a_id to column
    # b.a_id, which conflicts with relationship(s): 'BSub1.a' (copies a.id to b.a_id)
    # a_member = relationship('AMember')

# however, *no* warning should be emitted otherwise.


class BSub1(B):

    a = relationship('A')

    __mapper_args__ = {'polymorphic_identity': 'bsub1'}


class BSub2(B):

    a_member = relationship('AMember')

    @classmethod
    def __declare_first__(cls):
        cls.__table__.append_constraint(
            ForeignKeyConstraint(
                ('a_id', 'a_member_id'),
                ('a_member.a_id', 'a_member.a_member_id'))
        )

    __mapper_args__ = {'polymorphic_identity': 'bsub2'}


engine = create_engine('sqlite://', echo=True)
Base.metadata.create_all(engine)

Session = sessionmaker(bind=engine, autoflush=False)
session = Session()


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.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
print session.query(B, A).outerjoin(B.a).all()

patch:

diff --git a/lib/sqlalchemy/orm/mapper.py b/lib/sqlalchemy/orm/mapper.py
index 1d172f71a..edfe45030 100644
--- a/lib/sqlalchemy/orm/mapper.py
+++ b/lib/sqlalchemy/orm/mapper.py
@@ -2450,6 +2450,15 @@ class Mapper(InspectionAttr):
             m = m.inherits
         return bool(m)

+    def shares_lineage(self, other):
+        """Return True if either this mapper or given mapper inherit from the other.
+
+        This is a bidirectional form of "isa".
+
+        """
+
+        return self.isa(other) or other.isa(self)
+
     def iterate_to_root(self):
         m = self
         while m:
diff --git a/lib/sqlalchemy/orm/relationships.py b/lib/sqlalchemy/orm/relationships.py
index 94c0d6694..89ef641f8 100644
--- a/lib/sqlalchemy/orm/relationships.py
+++ b/lib/sqlalchemy/orm/relationships.py
@@ -2693,6 +2693,7 @@ class JoinCondition(object):
                 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.parent.shares_lineage(pr.parent) and \
                         fr_ is not from_ and \
                             pr not in self.prop._reverse_property:
                         other_props.append((pr, fr_))

Comments (2)