SQLAlchemy 2.x：如何预加载关联集合并筛选指定子对象的父对象

背景 #

在SQLAlchemy 2.x中，如何预加载关联集合？

假设我们有如下Parent和Child模型：

class Parent(Base):
    __tablename__ = "parent"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(30))

    children: Mapped[List["Child"]] = relationship(
        back_populates="parent", cascade="all, delete-orphan"
    )

    def __repr__(self) -> str:
        return f"Parent(id={self.id!r}, name={self.name!r})"

class Child(Base):
    __tablename__ = "child"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(30))

    parent_id: Mapped[int] = mapped_column(ForeignKey("parent.id"))
    parent: Mapped["Parent"] = relationship(back_populates="children")

    def __repr__(self) -> str:
        return f"Child(id={self.id!r}, name={self.name!r})"

我们需要获取拥有id为1的Child的Parent对象，并将结果中的parent.children填充为该父对象的所有子对象。

现有数据如下：
parent表

id  name
1   p1
2   p2

child表

id  name  parent_id
1   c1    1
2   c2    1
3   c3    1
4   c4    1
5   c5    2

期望查询结果：

result = <查询拥有id==1的子对象的父对象>
print(result)
>>> Parent(id=1, name='p1')

print(result.children)
>>> [
        Child(id=1, name='c1'),
        Child(id=2, name='c2'),
        Child(id=3, name='c3'),
        Child(id=4, name='c4'),
    ]

错误的尝试 #

测试用例1：未预加载关联集合 #

stmt = select(Parent).join(Parent.children).where(Child.id == 1)

生成的SQL：

SELECT parent.id, parent.name 
FROM parent JOIN child ON parent.id = child.parent_id 
WHERE child.id = 1

该SQL能正确查询到目标父对象，但由于未指定预加载children，访问parent.children时会触发延迟加载，在异步环境下会抛出错误：

sqlalchemy.exc.MissingGreenlet: greenlet_spawn has not been called; can't call await_only() here.

测试用例2：错误使用joinedload导致笛卡尔积 #

stmt = select(Parent).options(joinedload(Parent.children)).where(Child.id == 1)

生成的SQL：

SELECT parent.id, parent.name, child_1.id AS id_1, child_1.name AS name_1, child_1.parent_id 
FROM parent JOIN child AS child_1 ON parent.id = child_1.parent_id, child 
WHERE child.id = 1

这里child表被直接放入FROM子句，形成了笛卡尔积，不符合需求。

正确解法 #

要实现需求，需要先通过子对象过滤父对象，同时预加载该父对象的所有子对象，可以通过以下两种方式实现：

方式1：使用contains_eager配合显式join #

contains_eager用于告诉SQLAlchemy，关联集合已经通过显式的JOIN加载完成，不需要再额外查询：

from sqlalchemy.orm import contains_eager

stmt = (
    select(Parent)
    .join(Parent.children)
    .options(contains_eager(Parent.children))
    .where(Child.id == 1)
    .distinct()  # 避免父对象因多子JOIN而重复
)

生成的SQL：

SELECT DISTINCT parent.id, parent.name, child.id AS id_1, child.name AS name_1, child.parent_id 
FROM parent JOIN child ON parent.id = child.parent_id 
WHERE child.id = 1

distinct()保证我们只获取唯一的父对象，同时contains_eager会将所有关联的子对象加载到parent.children中。

方式2：使用subquery过滤父对象，再预加载 #

先通过子查询找到目标父对象的ID，再查询父对象并预加载其所有子对象：

from sqlalchemy.orm import joinedload

subq = select(Child.parent_id).where(Child.id == 1).scalar_subquery()
stmt = select(Parent).options(joinedload(Parent.children)).where(Parent.id == subq)

生成的SQL：

SELECT parent.id, parent.name, child_1.id AS id_1, child_1.name AS name_1, child_1.parent_id 
FROM parent LEFT OUTER JOIN child AS child_1 ON parent.id = child_1.parent_id 
WHERE parent.id = (SELECT child.parent_id FROM child WHERE child.id = 1)

这种方式不会产生重复的父对象，也能完整预加载所有子对象，适合更复杂的过滤场景。

内容的提问来源于stack exchange，提问作者gmagno