Question: Common Table Expressions with SQLAlchemy

cte.py

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import Session, relationship, backref, sessionmaker
from sqlalchemy import create_engine, Column, String, Integer
from sqlalchemy.schema import ForeignKey

Base = declarative_base()


class Object(Base):
    """ Object ORM """
    __tablename__ = 'object'

    type = Column(String(30))
    __mapper_args__ = {
        'polymorphic_identity': 'object',
        'polymorphic_on': type
    }

    id = Column(Integer, primary_key=True)
    name = Column(String(256))

    def __repr__(self):
        return "<Object(id='%s', name='%s')>" % (self.id, self.name)


class A(Object):
    __tablename__ = 'a'

    __mapper_args__ = {
        'polymorphic_identity': 'a',
    }

    id = Column(Integer, ForeignKey('object.id'), primary_key=True)

    b_list = relationship(
        'B',
        secondary='a_b_association',
        backref='a_list'
    )


class B(Object):
    __tablename__ = 'b'

    __mapper_args__ = {
        'polymorphic_identity': 'b',
    }

    id = Column(Integer, ForeignKey('object.id'), primary_key=True)

    c_list = relationship(
        'C',
        secondary='b_c_association',
        backref='b_list'
    )


class ABAssociation(Base):
    __tablename__ = 'a_b_association'

    a_id = Column(Integer, ForeignKey('a.id'), primary_key=True)
    b_id = Column(Integer, ForeignKey('b.id'), primary_key=True)


class C(Object):
    __tablename__ = 'c'

    __mapper_args__ = {
        'polymorphic_identity': 'c',
    }

    id = Column(Integer, ForeignKey('object.id'), primary_key=True)


class BCAssociation(Base):
    __tablename__ = 'b_c_association'

    b_id = Column(Integer, ForeignKey('b.id'), primary_key=True)
    c_id = Column(Integer, ForeignKey('c.id'), primary_key=True)


class X(Base):
    __tablename__ = 'x'

    id = Column(Integer, primary_key=True)
    name = Column(String(30))

    obj_id = Column(Integer, ForeignKey('object.id'))
    obj = relationship('Object', backref='x_list')

    def __repr__(self):
        return "<X(id='%s', name='%s')>" % (self.id, self.name)


engine = create_engine("postgresql://questions_user:questions_pass@localhost:5432/questions_db", echo=True)

Base.metadata.drop_all(engine)
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()

a = A(name='a1')
session.add(a)
b = B(name='b1', a_list=[a])
b2 = B(name='b2')
session.add(b)
session.add(b2)
c = C(name='c1', b_list=[b])
c2 = C(name='c2', b_list=[b2])
session.add(c)
session.add(c2)

x = X(name='x1', obj=a)
x = X(name='x1', obj=b2)
x = X(name='x1', obj=c)
session.add(x)

session.commit()


cte_c_bs = session.query(
        BCAssociation.b_id,
        BCAssociation.c_id
    ).filter(
        BCAssociation.c_id == 4
    ).cte()

cte_b_as = session.query(
        ABAssociation.a_id,
        cte_c_bs.c.c_id
    ).join(
        cte_c_bs,
        cte_c_bs.c.b_id == ABAssociation.b_id
    ).cte()


# SELECT x.obj_id AS c_id,
#        x.obj_id AS obj_id,
#        x.name AS name
# FROM x
# WHERE x.obj_id IN (4)
# AND x.name = 'x1'

q1 = session.query(
        X.obj_id,
        X.obj_id.label('somethingelse'),
        X.name
    ).filter(
        X.obj_id == 4,
        X.name == 'x1'
    )

q2 = session.query(
        cte_c_bs.c.c_id,
        cte_c_bs.c.b_id,
        X.name
    ).join(
        X,
        cte_c_bs.c.b_id == X.obj_id
    ).filter(
        X.name == 'x1'
    )

q3 = session.query(
        cte_b_as.c.c_id,
        cte_b_as.c.a_id,
        X.name
    ).join(
        X,
        cte_b_as.c.a_id == X.obj_id
    ).filter(
        X.name == 'x1'
    )

q = q1.union(q2).union(q3)

# SELECT cte_b_as.c_id AS c_id,
#        x.obj_id AS obj_id,
#        x.name AS name
# FROM cte_b_as
# INNER JOIN x ON cte_b_as.a_id = x.obj_id
# WHERE x.name = 'x1';


print q.all()

cte.sql

WITH cte_c_bs AS
(
    SELECT b_c_association.b_id AS b_id,
           b_c_association.c_id AS c_id
    FROM b_c_association
    WHERE b_c_association.c_id IN (5)
),

cte_b_as AS
(
    SELECT a_b_association.a_id AS a_id,
           cte_c_bs.c_id AS c_id
    FROM a_b_association
    INNER JOIN cte_c_bs ON a_b_association.b_id = cte_c_bs.b_id
)

-- SELECT x.obj_id AS c_id,
--        x.obj_id AS obj_id,
--        x.name AS name
-- FROM x
-- WHERE x.obj_id IN (4)
-- AND x.name = 'x1'

-- UNION

SELECT cte_c_bs.c_id AS c_id,
       x.obj_id AS obj_id,
       x.name AS name
FROM cte_c_bs
INNER JOIN x ON cte_c_bs.b_id = x.obj_id
WHERE x.name = 'x2'

-- UNION

-- SELECT cte_b_as.c_id AS c_id,
--        x.obj_id AS obj_id,
--        x.name AS name
-- FROM cte_b_as
-- INNER JOIN x ON cte_b_as.a_id = x.obj_id
-- WHERE x.name = 'x1';