SQLAlchemy polymorphic joinedload 1.0.10

test_sqla.py

import sqlalchemy as sa
import sqlalchemy.orm
from sqlalchemy.orm import joinedload
import sqlalchemy.ext.declarative
import sqlalchemy.ext.orderinglist

Base = sqlalchemy.ext.declarative.declarative_base()


class Person(Base):
    __tablename__ = 'person'

    id = sa.Column(sa.Integer, nullable=False, primary_key=True)
    person_type = sa.Column(sa.String(20), nullable=False)

    __mapper_args__ = {
        'polymorphic_on': person_type,
        'with_polymorphic': '*',
    }


class User(Person):
    __mapper_args__ = {'polymorphic_identity': 'user'}


class Admin(Person):
    __mapper_args__ = {'polymorphic_identity': 'admin'}


class Action(Base):
    __tablename__ = 'action'
    id = sa.Column(sa.Integer, nullable=False, primary_key=True)

    author_id = sa.Column(sa.Integer, sa.ForeignKey(User.id))
    author = sa.orm.relationship('User')

    post_type = sa.Column(sa.String(20), nullable=False)

    __mapper_args__ = {
        'polymorphic_on': post_type,
        'polymorphic_identity': 'post',
        'with_polymorphic': '*',
    }


class Like(Action):
    __tablename__ = 'like'
    __mapper_args__ = {'polymorphic_identity': 'like'}
    id = sa.Column(sa.Integer, sa.ForeignKey('action.id'), primary_key=True)


class Comment(Action):
    __tablename__ = 'comment'
    __mapper_args__ = {'polymorphic_identity': 'comment'}
    id = sa.Column(sa.Integer, sa.ForeignKey('action.id'), primary_key=True)

    article_id = sa.Column(sa.Integer, sa.ForeignKey('article.id'))
    article = sa.orm.relationship('Article', primaryjoin='Article.id==Comment.article_id', backref='comments')


class Article(Base):
    __tablename__ = 'article'
    id = sa.Column(sa.Integer, nullable=False, primary_key=True)


engine = sa.create_engine('sqlite:///:memory:', echo=True)

Base.metadata.drop_all(engine)
Base.metadata.create_all(engine)

Session = sa.orm.sessionmaker(bind=engine)
session = Session()

user1 = User(id=1)
admin1 = Admin(id=2)
user2 = User(id=3)
article1 = Article(id=1)
comment1 = Comment(id=2, author=user1, article_id=1)
comment2 = Comment(id=3, author=user2, article_id=1)

session.add_all([user1, user2, admin1, article1, comment1, comment2])
session.commit()
session.expunge_all()


# Works with 1.0.9
# Not working with 1.0.10: raise AssertionError
try:
    new_n1 = session.query(Article).options(
        joinedload(Article.comments).joinedload(Action.author),
    ).all()

    print new_n1
    print new_n1[0].comments
    print new_n1[0].comments[0].author
except AssertionError as e:
    print e


# Works with 1.0.9
# Not working with 1.0.10: 2 queries
try:
    new_n1 = session.query(Article).options(
        joinedload('comments'),
        joinedload('comments.author')
    ).all()

    print new_n1
    print new_n1[0].comments
    print new_n1[0].comments[0].author
except AssertionError as e:
    print e

# Works with 1.0.9
# Not working with 1.0.10: raise AssertionError
try:
    new_n1 = session.query(Article).options(
        joinedload('comments').joinedload('author')
    ).all()

    print new_n1
    print new_n1[0].comments
    print new_n1[0].comments[0].author
except AssertionError as e:
    print e

# Works with 1.0.9
# Not working with 1.0.10: raise AssertionError
try:
    new_n1 = session.query(Article).options(
        joinedload('comments').load_only('id', 'author_id').joinedload('author').load_only('id')
    ).all()

    print new_n1
    print new_n1[0].comments
    print new_n1[0].comments[0].author
except AssertionError as e:
    print e