Seth P
unread,Jun 14, 2013, 11:52:21 PM6/14/13Sign in to reply to author
Sign in to forward
You do not have permission to delete messages in this group
Either email addresses are anonymous for this group or you need the view member email addresses permission to view the original message
to sqlal...@googlegroups.com
I've encountered what I believe to be a bug in SQLAlchemy (versions 0.8.0 and 0.8.1) in a query that joins class/tables that use joined inheritance.
In the code below, I would expect the three queries to produce the same output, namely [u'CCC'], but the first one gives a different (incorrect) result, [u'BBB']. Is this a bug, or is the query malformed?
In the second query, adding a seemingly gratuitous join with D fixes the problem. And as the third query shows, replacing C by an aliased version also fixes the problem. So whatever is going on seems rather subtle.
Thanks,
Seth
from sqlalchemy import Column, Integer, String, ForeignKey, create_engine
from sqlalchemy.ext.declarative.api import declarative_base
from sqlalchemy.orm import relationship, sessionmaker, scoped_session
from sqlalchemy.orm.util import aliased
Base = declarative_base(object)
metadata = Base.metadata
class A(Base):
__tablename__ = 'A'
idx = Column(Integer, primary_key=True)
name = Column(String(20), nullable=False)
type_idx = Column(Integer, nullable=False)
__mapper_args__ = { 'polymorphic_on':type_idx }
class B(A):
__tablename__ = 'B'
idx = Column(Integer, ForeignKey(str(A.__table__) + ".idx"), primary_key=True)
__mapper_args__ = { 'polymorphic_identity':2 }
class C(A):
__tablename__ = 'C'
idx = Column(Integer, ForeignKey(str(A.__table__) + ".idx"), primary_key=True)
b_idx = Column(Integer, ForeignKey(str(B.__table__) + ".idx"), nullable=False)
b = relationship("B", foreign_keys=[b_idx])
__mapper_args__ = { 'polymorphic_identity':3 }
class D(A):
__tablename__ = 'D'
idx = Column(Integer, ForeignKey(str(A.__table__) + ".idx"), primary_key=True)
__mapper_args__ = { 'polymorphic_identity':4 }
class CtoD(Base):
__tablename__ = 'CtoD'
idx = Column(Integer, primary_key=True)
c_idx = Column(Integer, ForeignKey(str(C.__table__) + ".idx"), nullable=False)
c = relationship("C", foreign_keys=[c_idx])
d_idx = Column(Integer, ForeignKey(str(D.__table__) + ".idx"), nullable=False)
d = relationship("D", foreign_keys=[d_idx])
if __name__ == '__main__':
engine = create_engine('sqlite:///:memory:', echo=False)
metadata.create_all(bind=engine)
Session = scoped_session(sessionmaker(bind=engine))
session = Session()
# populate tables with a single entry in each table
b = B(name='BBB')
c = C(name='CCC', b=b)
d = D(name='DDD')
c_to_d = CtoD(c=c, d=d)
session.add_all([b, c, d, c_to_d])
session.commit()
sql_query = session.query(B, C.name).join(C, B.idx == C.b_idx).join(CtoD, C.idx == CtoD.c_idx).join(D, CtoD.d_idx == D.idx)
print [name for (_, name) in sql_query.all()] # [u'BBB']
sql_query = session.query(B, C.name).join(C, B.idx == C.b_idx).join(CtoD, C.idx == CtoD.c_idx)
print [name for (_, name) in sql_query.all()] # [u'CCC']
aliased_C = aliased(C)
sql_query = session.query(B, aliased_C.name).join(aliased_C, B.idx == aliased_C.b_idx).join(CtoD, aliased_C.idx == CtoD.c_idx).join(D, CtoD.d_idx == D.idx).join(D, CtoD.d_idx == D.idx)
print [name for (_, name) in sql_query.all()] # [u'CCC']