Bad KnowHow of SQLAlchemy

broken_ordering.py

# -*- coding: utf8 -*-

from __future__ import print_function
import sqlalchemy as sa
from sqlalchemy.orm import backref, joinedload, relationship, subqueryload
from sqlalchemy.ext.declarative import declarative_base


Base = declarative_base()


class Tag(Base):
    __tablename__ = 'tag'
    id = sa.Column(sa.Integer, primary_key=True, autoincrement=True)
    ordering = sa.Column(sa.Integer)


class Resource(Base):
    __tablename__ = 'resource'
    id = sa.Column(sa.Integer, primary_key=True, autoincrement=True)


class ResourceTag(Base):
    __tablename__ = 'resource_tag'
    resource_id = sa.Column(sa.ForeignKey(Resource.id), primary_key=True)
    tag_id = sa.Column(sa.ForeignKey(Tag.id), primary_key=True)
    # ここでorder_byに別表への参照を入れるとアウト。
    resource = relationship(Resource, backref=backref('tag_maps', order_by=Tag.ordering), uselist=False)
    tag = relationship(Tag, backref='resource_maps', uselist=False)


def main(url):
    from sqlalchemy.orm import sessionmaker
    engine = sa.create_engine(url)
    Base.metadata.create_all(bind=engine)
    sm = sessionmaker()
    sm.configure(bind=engine)
    db = sm()
    db.add(Resource())
    db.flush()
    rs = db.query(Resource).options(
        subqueryload(Resource.tag_maps).joinedload(ResourceTag.tag),
    ).all()
    print(rs)


if __name__ == '__main__':
    import sys
    url = 'sqlite://' if len(sys.argv) < 2 else sys.argv[1]
    main(url)

"""
ResourceTag.resource の backref で ``Tag.ordering`` を書き込んでいるので別名 ``tag_1`` が使われず、SQLエラーになる
sqlalchemy.exc.OperationalError: (sqlite3.OperationalError) no such column: tag.ordering [SQL: 'SELECT resource_tag.resource_id AS resource_tag_resource_id, resource_tag.tag_id AS resource_tag_tag_id, anon_1.resource_id AS anon_1_resource_id, tag_1.id AS tag_1_id, tag_1.ordering AS tag_1_ordering \nFROM (SELECT resource.id AS resource_id \nFROM resource) AS anon_1 JOIN resource_tag ON anon_1.resource_id = resource_tag.resource_id LEFT OUTER JOIN tag AS tag_1 ON tag_1.id = resource_tag.tag_id ORDER BY anon_1.resource_id, tag.ordering']
"""

broken_ordering_workaround.py

# -*- coding: utf8 -*-

from __future__ import print_function
import sqlalchemy as sa
from sqlalchemy.orm import relationship, subqueryload
from sqlalchemy.ext.declarative import declarative_base


Base = declarative_base()


class Tag(Base):
    __tablename__ = 'tag'
    id = sa.Column(sa.Integer, primary_key=True, autoincrement=True)
    ordering = sa.Column(sa.Integer)

    def __repr__(self):
        return '<Tag id=%d, ordering=%d>' % (self.id, self.ordering)


class Resource(Base):
    __tablename__ = 'resource'
    id = sa.Column(sa.Integer, primary_key=True, autoincrement=True)

    # 直接Relationを作っちゃうことでワークアラウンド
    tags = relationship(
        Tag,
        primaryjoin='Resource.id == foreign(resource_tag.c.resource_id)',
        secondary='resource_tag',
        secondaryjoin='foreign(resource_tag.c.tag_id) == Tag.id',
        order_by='Tag.ordering',
    )

    def __repr__(self):
        return '<Resource id=%d, tags=%r>' % (self.id, self.tags)


class ResourceTag(Base):
    __tablename__ = 'resource_tag'
    resource_id = sa.Column(sa.Integer, sa.ForeignKey(Resource.id), primary_key=True)
    tag_id = sa.Column(sa.Integer, sa.ForeignKey(Tag.id), primary_key=True)
    resource = relationship(Resource, backref='tag_maps', uselist=False)
    tag = relationship(Tag, backref='resource_maps', uselist=False)


def main(url):
    from sqlalchemy.orm import sessionmaker
    engine = sa.create_engine(url, echo=True)
    Base.metadata.create_all(bind=engine)
    sm = sessionmaker()
    sm.configure(bind=engine)
    db = sm()
    db.add(Resource(id=1))
    db.add(Tag(id=1, ordering=100))
    db.add(Tag(id=2, ordering=99))
    db.flush()
    db.add(ResourceTag(resource_id=1, tag_id=1))
    db.add(ResourceTag(resource_id=1, tag_id=2))
    db.flush()
    rs = db.query(Resource).options(
        subqueryload(Resource.tags)
    ).order_by(Resource.id).all()
    print(rs)


if __name__ == '__main__':
    import sys
    url = 'sqlite://' if len(sys.argv) < 2 else sys.argv[1]
    main(url)