如何在SQLAlchemy中循环链式调用selectinload()？

问题场景 #

现有手动编写的SQLAlchemy查询如下：

statement = select(Parent)
    .filter(Parent.id == parent_id)
    .options(selectinload(Parent.child)
             .selectinload(Child.nodes)
             .selectinload(Child.nodes)
             .selectinload(Child.nodes)
    )
result = await async_session.execute(statement)

parent = result.scalars().first()

由于Child.nodes是自引用一对多关系，需要根据指定的深度（例如6级）自动循环链式调用selectinload()，避免手动重复编写多层嵌套的代码。

现有实现的局限 #

目前已实现通过循环添加独立的options项：

select_options = [
    selectinload(Parent.attr1),
    selectinload(Parent.attr2),
    selectinload(Parent.child).selectinload(Child.nodes).selectinload(Child.nodes),
]
query = select(Model)
query = query.filter(Model.id == model_id)
for option in select_options:
    query = query.options(option)

result = await async_session.execute(query)

但这种方式仅能添加多个独立的加载选项，无法在单个选项内部实现多级嵌套的selectinload循环调用。

解决方案：动态构建链式加载选项 #

可以编写一个工具函数，从基础加载项开始，循环指定次数来链式调用selectinload，生成所需的多级嵌套加载选项：

from sqlalchemy import select
from sqlalchemy.orm import selectinload

def build_nested_selectinload(base_load, relation, depth):
    current_load = base_load
    for _ in range(depth):
        current_load = current_load.selectinload(relation)
    return current_load

# 使用示例：构建Parent.child -> Child.nodes 6级嵌套加载
target_depth = 6
# 初始加载Parent.child，然后链式添加6次Child.nodes的selectinload
nested_load = build_nested_selectinload(selectinload(Parent.child), Child.nodes, target_depth)

# 组装最终查询
statement = select(Parent)\
    .filter(Parent.id == parent_id)\
    .options(
        selectinload(Parent.attr1),
        selectinload(Parent.attr2),
        nested_load  # 加入动态生成的多级嵌套加载项
    )

result = await async_session.execute(statement)
parent = result.scalars().first()

环境信息 #

• SQLAlchemy 2.0.36
• asyncpg 0.30.0
• fastapi 0.112.0

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