sqlalchemy contains_eager 不加载嵌套关系

Question

我有 motor 和 motor_cycle 表，这是一对多的关系。我想加入这些表并显示 motor_cycle 表上具有 OPTIONAL 条件的电机的所有 motor_cycle。最初我使用的是joinedload，但发现应用where条件后它只会返回所有记录。然后我切换到 contains_eager 但有一些不寻常的行为。当我传入过滤条件时，它似乎正确应用并返回一个 motor_cycle 记录。但是，当我不传递任何条件时，它不会返回所有 motor_cycle 记录，而是只返回一个。它返回的似乎每次都是一样的，并且与我之前使用的条件不匹配（但这并不重要）。我可以采取什么不同的方式将所有 motor_cycle 记录加载到我的响应中，但同时能够选择过滤它们。

编辑：我已将问题范围缩小到所应用的限制。发送请求时，我传入的限制为 1。但是，此限制应应用于父对象而不是子对象。不是将 1 的限制应用于 motor，而是将其应用于 motor_cycle。知道如何解决这个问题吗？

async def search_motors(
    session: AsyncSession,
    serial_number: str = None,
    is_latest_fl: str = None,
    limit: int = 20,
    offset: int = 0,
):
    async with session:

        statement = (
            select(
                DBMotor,
            ).options(contains_eager(DBMotor.motor_cycle))
            .join(DBMotor.motor_cycle)
        )

        if serial_number:
            statement = statement.where(
                DBMotor.serial_number.ilike(f"%{serial_number}%")
            )

        if is_latest_fl:
            statement = statement.where(DBMotorCycle.is_latest_fl == is_latest_fl)

     

        statement = statement.limit(limit).offset(offset)
        result = await session.scalars(statement)
        list_of_motors = result.unique().all()
        list_of_motor_models = (
            [MotorSearchModel.model_validate(motor) for motor in list_of_motors]
            if list_of_motors
            else None
        )

    return list_of_motor_models

Answer 1

问题不在于 eager_loading，而在于限制以及它在语句中的应用方式。该限制没有应用于电机的父对象，而是应用于子电机。

我使用子查询首先获取应用了限制的电机记录，然后使用它来连接到查询的其余部分。以下是最终代码。

async def search_motors(
    session: AsyncSession,
    serial_number: str = None,
    is_latest_fl: str = None,
    limit: int = 20,
    offset: int = 0,
):
    async with session:
        
        subquery = select(DBMotor).limit(limit).offset(offset).subquery()
        statement = (
            select(
                DBMotor,
            ).options(contains_eager(DBMotor.motor_cycle))
            .join(DBMotor.motor_cycle).outerjoin(subquery, DBMotor.motor_id == subquery.c.motor_id)
        )

        if serial_number:
            statement = statement.where(
                DBMotor.serial_number.ilike(f"%{serial_number}%")
            )

        if is_latest_fl:
            statement = statement.where(DBMotorCycle.is_latest_fl == is_latest_fl)

     

        result = await session.scalars(statement)
        list_of_motors = result.unique().all()
        list_of_motor_models = (
            [MotorSearchModel.model_validate(motor) for motor in list_of_motors]
            if list_of_motors
            else None
        )

    return list_of_motor_models