SQLAlchemy 2.x中如何限制每个QuizTopic仅返回3条关联Quiz记录

问题分析 #

当前用joinedload加载关联数据时，会拉取每个QuizTopic下的全部Quiz记录。SQLAlchemy的常规关联加载不支持直接给单组关联数据设数量限制，得靠子查询或窗口函数实现「按topic_id分组取前3条」的逻辑。

解决方案1：用窗口函数子查询筛选目标记录 #

通过子查询先拿到每个topic_id对应的前3条Quiz的ID，再关联查询只加载这些匹配的记录，适合一次性批量加载场景：

修改orm.py中的查询代码：

async def get_content_topics(session: AsyncSession, limit: int, continue_after: int):
    # 子查询：给每个topic下的quiz编号，取前3条的id
    quiz_subquery = (
        select(
            Quiz.id,
            # 按topic_id分组，按quiz的id排序（可换成你需要的排序字段，比如创建时间）
            func.row_number().over(
                partition_by=Quiz.topic_id,
                order_by=Quiz.id
            ).label("row_num")
        )
    ).subquery()

    smt = (
        select(QuizTopic)
        .order_by(QuizTopic.id)
        .offset(continue_after).limit(limit)
        .options(
            load_only(QuizTopic.id, QuizTopic.title),
            # 用contains_eager加载筛选后的quiz
            contains_eager(QuizTopic.quizzes_info).load_only(
                Quiz.id, Quiz.logo_url, Quiz.title, Quiz.meta
            )
        )
        .join(QuizTopic.quizzes_info)
        .join(quiz_subquery, Quiz.id == quiz_subquery.c.id)
        .filter(quiz_subquery.c.row_num <= 3)
    )
    result = await session.scalars(smt)
    return result.unique().all()

解决方案2：动态加载关联数据（按需查询） #

如果不需要一次性加载所有关联数据，可以修改关系的加载模式，在拿到QuizTopic后单独查询前3条Quiz，适合小批量topic场景：

首先修改models.py中的关系定义：

class QuizTopic(Base):
    __tablename__ = "quiz_topic"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    title: Mapped[str] = mapped_column(String(60))

    quizzes_info: Mapped[list["Quiz"]] = relationship(
        back_populates="topic_info",
        cascade="all, delete",
        passive_deletes=True,
        lazy='dynamic'  # 改为动态加载，返回Query对象而非直接返回列表
    )

然后修改查询函数：

async def get_content_topics(session: AsyncSession, limit: int, continue_after: int):
    smt = (
        select(QuizTopic)
        .order_by(QuizTopic.id)
        .offset(continue_after).limit(limit)
        .options(load_only(QuizTopic.id, QuizTopic.title))
    )
    topics = await session.scalars(smt)
    topics_list = topics.unique().all()
    
    # 给每个topic单独取前3条quiz
    for topic in topics_list:
        topic.quizzes_info = await topic.quizzes_info.limit(3).options(
            load_only(Quiz.id, Quiz.logo_url, Quiz.title, Quiz.meta)
        ).all()
    return topics_list

注意事项 #

• 方案1的窗口函数需要数据库支持（如PostgreSQL、MySQL 8.0+），旧版本数据库需用其他方式实现分组取前N条。
• 方案2会产生N+1查询（每个topic对应一条查询），topic数量多时性能会受影响，谨慎使用。
• 排序字段可根据业务需求调整，比如换成Quiz.create_time.desc()来取最新的3条记录。

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