如何根据子对象中的关系在 SQLAlchemy 中创建条件关系？

Question

我正在尝试在 父 对象上创建一个名为 children_with_grandchild 的条件关系，该对象仅包含具有至少一个 子 对象>孙子。我有一个这样的模型：

from sqlalchemy import Column, ForeignKey, Integer, create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship

engine = create_engine('sqlite:///:memory:', echo=True)
Base = declarative_base()
Session = sessionmaker(bind=engine)


def setup():
    Base.metadata.create_all(engine)


class Parent(Base):
    __tablename__ = 'parent'

    id = Column(Integer, primary_key=True)

    children = relationship('Child', lazy='joined')


class Child(Base):
    __tablename__ = 'child'

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

    grandchildren = relationship('GrandChild', lazy='joined')


class GrandChild(Base):
    __tablename__ = 'grandchild'

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

到目前为止我尝试过但没有奏效的方法是将以下内容添加到 Parent 类中：

children_with_grandchild = relationship('Child',
                                        primaryjoin='and_(Parent.id==Child.parent_id,'
                                                    'Child.grandchildern.any())',
                                        lazy='joined')

根据我在尝试此操作时收到的错误消息，似乎只允许在 primaryjoin 属性的条件中使用列，而不是关系。所以我也尝试将以下内容添加到 Parent 类中：

children_with_grandchild = relationship('Child',
                                        primaryjoin='and_(Parent.id==Child.parent_id,'
                                                    'Child.id==Grandchild.child_id)',
                                        lazy='joined')

但是，当我查询 Parent 记录时出现的错误是：

sqlalchemy.exc.OperationalError: (sqlite3.OperationalError) 没有这样的列：grandchild.child_id

此错误似乎是由包含 lazy='joined' 属性引起的，因为如果我删除它，它可以正常工作。问题是查询优化我真的需要在初始查询中加载 childern_with_grandchild 关系。

给定以下测试数据，关系 children_with_grandchild 应该只包含 c1 孩子，而关系 children 将同时包含 c1 和 c2 子对象。

>>> setup()
>>> session = Session()
>>> p = Parent(id=1)
>>> c1 = Child(id=1, parent_id=1)
>>> c2 = Child(id=2, parent_id=1)
>>> gc1 = GrandChild(id=1, child_id=1)
>>> session.add_all([p, c1, c2, gc1])
>>> session.commit()
>>> p = session.query(Parent).first()

任何帮助将不胜感激。谢谢！

Answer 1

我会尝试这样的事情

在parenttable 上添加

@property
def has_grandchildren(self):
    return any(child.grandchildren for child in self.children)

然后您可以通过某种形式获得所有 parents_with_no_grandchildren 的列表

parents = session.query(Parent).all()
parents_with_no_grandchildren = [parent for parent in parents if not parent.has_grandchildren]