On May 25, 2013, at 10:56 AM, Michael Bayer <[email protected]> wrote:

> 
> 
> SQLAlchemy can do it, if you set up a full "with_polymorphic" join of all the 
> tables, then use a CASE statement against all the tables to determine the 
> discriminator.

Here's that (I'll admit I'm surprised it works):

from sqlalchemy import *
from sqlalchemy.orm import *
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class A(Base):
    __tablename__ = 'a'

    id = Column(Integer, primary_key=True)

    def __repr__(self):
        return "A()"

    __mapper_args__ = {"polymorphic_identity": "a"}

class B(A):
    __tablename__ = 'b'

    id = Column(Integer, ForeignKey('a.id'), primary_key=True)
    b_data = Column(String)

    def __repr__(self):
        return "B(b_data=%r)" % self.b_data

    __mapper_args__ = {"polymorphic_identity": "b"}

class C(A):
    __tablename__ = 'c'

    id = Column(Integer, ForeignKey('a.id'), primary_key=True)
    c_data = Column(String)

    def __repr__(self):
        return "C(c_data=%r)" % self.c_data

    __mapper_args__ = {"polymorphic_identity": "c"}

class D(A):
    __tablename__ = 'd'

    id = Column(Integer, ForeignKey('a.id'), primary_key=True)
    d_data = Column(String)

    def __repr__(self):
        return "D(d_data=%r)" % self.d_data

    __mapper_args__ = {"polymorphic_identity": "d"}


# left outer join all the tables
pu = select([A.__table__, B.__table__, C.__table__, D.__table__,
                case([(B.id != None, "b"), (C.id != None, "c"),
                        (D.id != None, "d")], else_=literal("a")).label('type')
            ], use_labels=True).select_from(
            A.__table__.outerjoin(B.__table__).
                outerjoin(C.__table__).outerjoin(D.__table__)
            ).alias()

# use that as polymorphic_on, discriminator
A.__mapper__.with_polymorphic = ('*', pu)
A.__mapper__.polymorphic_on = pu.c.type
A.type = column_property(pu.c.type)


e = create_engine("sqlite://", echo=True)
Base.metadata.create_all(e)

sess = Session(e)
sess.add_all([B(b_data="b1"), C(c_data="c1"), D(d_data="d1"),
                    A(), B(b_data="b2"), C(c_data="c2")])
sess.commit()
sess.close()

sess = Session(e)
print sess.query(A).all()

-- 
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?hl=en.
For more options, visit https://groups.google.com/groups/opt_out.


Reply via email to