Sqlalchemy：查询多态类的 m2m 关系

Question

我有两个通过多对多关系连接的类：Parent 和 Tag。

Base = declarative_base()

association_table = Table('associations', Base.metadata,
    Column('parent_id', Integer, ForeignKey('parent.id')),
    Column('tag_id', Integer, ForeignKey('tag.id')),
)

class Tag(Base):
    __tablename__ = 'tags'
    id = Column(Integer, Sequence('tag_id_seq'), primary_key=True)
    name = Column(String)


class Parent(Base):
    __tablename__ = 'parents'
    id = Column(Integer, Sequence('parent_id_seq'), primary_key=True)
    tags = relationship('Tag', secondary=association_table, backref='parents')

如果我想查询与Parent 有一个或多个关系的所有Tag 对象，我会这样做：

session.query(Tag).filter(Tag.parents.any()).all()

但是，这个Parent 类是子类Alice 和Bob 的父类：

class Alice(Parent):
    __tablename__ = 'alices'
    __mapper_args__ = {'polymorphic_identity': 'alice'}
    alice_id = Column(Integer, ForeignKey('parents.id'), primary_key=True)

class Bob(Parent):
    __tablename__ = 'bobs'
    __mapper_args__ = {'polymorphic_identity': 'bob'}
    bob_id = Column(Integer, ForeignKey('parents.id'), primary_key=True)

现在我希望能够检索与Alice 对象具有一个或多个关系的所有Tag 对象。前面的查询 session.query(Tag).filter(Tag.parents.any()).all() 不会这样做，因为它不区分 Alice 或 Bob 对象 - 它甚至不知道它们的存在。

我已经搞砸了 Query 一段时间但没有成功，所以我假设如果可以完成，它必须与 Table 类中的一些额外代码行有关，如上所示。虽然文档包含有关 polymorphic classes 和 many-to-many relations 的信息，并且 Mike Bayer 本人向某人 this 提供了一个看似相关的问题的答案，这个问题看起来很有趣，但我还远远不能理解，但我有点卡住了。

这些代码示例可能会让 Python 解释器感到厌恶，但希望能理解我的意思。给那些可以在我的路上帮助我的人！

Answer 1

在编写一点 MWE 时，我实际上找到了一个解决方案，这实际上与我已经尝试过的几乎相同。不过，budder 给了我新的希望，谢谢 :)

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


Base = declarative_base()

association_table = Table('associations', Base.metadata,
    Column('parent_id', Integer, ForeignKey('parents.id')),
    Column('tag_id', Integer, ForeignKey('tags.id')),
)


class Tag(Base):
    __tablename__ = 'tags'
    id = Column(Integer, Sequence('tag_id_seq'), primary_key=True)
    name = Column(String)


class Parent(Base):
    __tablename__ = 'parents'
    id = Column(Integer, Sequence('parent_id_seq'), primary_key=True)
    tags = relationship('Tag', secondary=association_table, backref='parents')


class Alice(Parent):
    __tablename__ = 'alices'
    __mapper_args__ = {'polymorphic_identity': 'alice'}
    alice_id = Column(Integer, ForeignKey('parents.id'), primary_key=True)


class Bob(Parent):
    __tablename__ = 'bobs'
    __mapper_args__ = {'polymorphic_identity': 'bob'}
    bob_id = Column(Integer, ForeignKey('parents.id'), primary_key=True)


engine = create_engine("sqlite://")
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()

tag_a = Tag(name='a')
tag_b = Tag(name='b')
tag_c = Tag(name='c')
session.add(tag_a)
session.add(tag_b)
session.add(tag_c)
session.commit()
session.add(Alice(tags=[tag_a]))
session.add(Bob(tags=[tag_b]))
session.commit()

for tag in session.query(Tag).\
        filter(Tag.parents.any(Parent.id == Alice.alice_id)).\
        all():
    print(tag.name)

如果有更好的替代方法，我仍然会感兴趣。我可以想象 sqlalchemy 提供了一种更直接、更优雅的方法，例如：

for tag in session.query(Tag).\
        filter(Tag.alices.any()).\
        all():
    print(tag.name)

Answer 2

如果您以某种方式编写对象类，则可以使用钝力方法...只需搜索它们...有点像粗略的查询方法：

all_tag_objects = session.query(Tag).all()  ## All tag objects in your database
tags = []
for tag in all_tag_objects:
    for parent in tag.parents:
        if parent.alices != []:  ## If there are alice objects in the tag parents alice reltionship instance variable...Then we append the tag because it meets our criteria.
             flagged_tags.append(tag)

听起来您找到了更好的方法，但我想最终的测试将是实际进行速度测试。