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']
--
You received this message because you are subscribed to the Google Groups
"sqlalchemy" group.
To unsubscribe from this group and stop receiving emails from it, send an email
to [email protected].
To post to this group, send email to [email protected].
Visit this group at http://groups.google.com/group/sqlalchemy.
For more options, visit https://groups.google.com/groups/opt_out.
