SQLAlchemy：执行GROUP BY与SUM时如何使用joinedload加载关联对象

问题场景 #

查询DBServiceOrder主对象时，通过joinedload关联加载DBVendor，同时对关联的DBServiceOrderItem的unit_price执行SUM聚合并按DBServiceOrder分组，执行后出现两个错误：

1. ORA-00979: not a GROUP BY expression：joinedload自动将DBVendor的字段加入SELECT，但这些字段未出现在GROUP BY中，Oracle要求所有非聚合字段必须在GROUP BY里。
2. ORA-00904: "VENDOR"."UPDATED_DATE": invalid identifier：手动添加DBVendor到GROUP BY后，SQLAlchemy生成的GROUP BY语句使用了vendor作为表别名，但实际查询中表的别名是vendor_1，导致字段引用错误。

用户原始代码：

async def search_service_orders(
    session: AsyncSession,
    service_order_number: str = None,
    source_service_order_number: str = None,
    limit: int = 20,
    offset: int = 0,
) -> Sequence[ServiceOrderSearchModel]:
    async with session:
        statement = (
            select(
                DBServiceOrder,
                func.sum(DBServiceOrderItem.unit_price).label("total_price"),
            )
            .options(joinedload(DBServiceOrder.vendor))
            .outerjoin(
                DBServiceOrderItem,
                DBServiceOrder.service_order_id == DBServiceOrderItem.service_order_id,
            )
            .group_by(DBServiceOrder)
        )

        if service_order_number is not None:
            statement = statement.where(
                DBServiceOrder.service_order_number.ilike(f"%{service_order_number}%")
            )
        if source_service_order_number is not None:
            statement = statement.where(
                DBServiceOrder.source_service_order_number.ilike(
                    f"%{source_service_order_number}%"
                )
            )

        statement = statement.limit(limit).offset(offset)
        result = await session.execute(statement)
        list_of_service_orders = result.all()

解决方案 #

方案一：先聚合再关联（推荐） #

通过子查询预先计算每个service_order的总价，再关联主表和vendor，避免GROUP BY包含大量非必要字段，同时规避别名问题。

async def search_service_orders(
    session: AsyncSession,
    service_order_number: str = None,
    source_service_order_number: str = None,
    limit: int = 20,
    offset: int = 0,
) -> Sequence[ServiceOrderSearchModel]:
    async with session:
        # 子查询计算每个服务订单的总价
        total_price_subq = (
            select(
                DBServiceOrderItem.service_order_id,
                func.sum(DBServiceOrderItem.unit_price).label("total_price")
            )
            .group_by(DBServiceOrderItem.service_order_id)
            .subquery()
        )
        
        # 主查询：加载服务订单、关联vendor、关联子查询结果
        statement = (
            select(DBServiceOrder, total_price_subq.c.total_price)
            .options(joinedload(DBServiceOrder.vendor))
            .outerjoin(total_price_subq, DBServiceOrder.service_order_id == total_price_subq.c.service_order_id)
        )

        # 过滤条件
        if service_order_number is not None:
            statement = statement.where(
                DBServiceOrder.service_order_number.ilike(f"%{service_order_number}%")
            )
        if source_service_order_number is not None:
            statement = statement.where(
                DBServiceOrder.source_service_order_number.ilike(
                    f"%{source_service_order_number}%"
                )
            )

        # 分页
        statement = statement.limit(limit).offset(offset)
        result = await session.execute(statement)
        list_of_service_orders = result.all()

方案二：修正GROUP BY的表别名 #

如果需要在主查询中完成聚合，需显式指定DBVendor的别名，确保GROUP BY语句使用正确的表别名引用字段。

async def search_service_orders(
    session: AsyncSession,
    service_order_number: str = None,
    source_service_order_number: str = None,
    limit: int = 20,
    offset: int = 0,
) -> Sequence[ServiceOrderSearchModel]:
    async with session:
        # 显式定义vendor的别名，确保GROUP BY和SELECT使用一致的别名
        vendor_alias = aliased(DBVendor)
        
        statement = (
            select(
                DBServiceOrder,
                func.sum(DBServiceOrderItem.unit_price).label("total_price"),
                vendor_alias
            )
            # 关联加载时指定别名
            .options(joinedload(DBServiceOrder.vendor, alias=vendor_alias))
            .outerjoin(DBServiceOrderItem, DBServiceOrder.service_order_id == DBServiceOrderItem.service_order_id)
            .outerjoin(vendor_alias, DBServiceOrder.vendor_id == vendor_alias.vendor_id)
            # 按主对象和vendor别名分组
            .group_by(DBServiceOrder, vendor_alias)
        )

        # 过滤条件
        if service_order_number is not None:
            statement = statement.where(
                DBServiceOrder.service_order_number.ilike(f"%{service_order_number}%")
            )
        if source_service_order_number is not None:
            statement = statement.where(
                DBServiceOrder.source_service_order_number.ilike(
                    f"%{source_service_order_number}%"
                )
            )

        # 分页
        statement = statement.limit(limit).offset(offset)
        result = await session.execute(statement)
        list_of_service_orders = result.all()

方案对比 #

• 方案一：性能更优，子查询仅聚合必要字段，主查询逻辑清晰，避免GROUP BY包含大量冗余字段，适合数据量较大的场景。
• 方案二：逻辑直接，但GROUP BY需要包含DBServiceOrder和DBVendor的所有字段，Oracle查询效率较低，仅适合简单场景。

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