如何将SqlAlchemy的group_by/func聚合查询转换为GraphQL查询

你现在的SQLAlchemy聚合查询返回的是字段值组成的元组（比如报错信息里的('dave', 1, Decimal('11.00'))），而你定义的Graphene类型期望接收的是带对应属性的对象，Graphene无法直接把元组映射到类型字段，因此会抛出不兼容实例的错误，同时返回空数据。

1. 定义匹配聚合结果的Graphene对象类型 #

专门定义一个承载客户统计数据的类型，不需要绑定原有表模型，灵活适配聚合结果：

import graphene

class CafeCustomerStats(graphene.ObjectType):
    username = graphene.String()
    order_count = graphene.Int(description="到店总次数")
    total_spent = graphene.Float(description="总消费金额")
    avg_spent_per_order = graphene.Float(description="单次到店平均消费")

    # 平均消费可以直接在类型层计算，不用修改SQL查询
    def resolve_avg_spent_per_order(parent, info):
        if parent.order_count == 0:
            return 0
        return float(parent.total_spent / parent.order_count)

2. 修改查询逻辑，把元组转换为Graphene类型实例 #

首先给聚合字段加label方便取值，再把查询返回的元组批量转成上面定义的类型实例：

# 定义查询字段，指定返回类型为CafeCustomerStats的列表
customers_by_cafe = graphene.List(CafeCustomerStats, token=graphene.String())

def resolve_customers_by_cafe(root, info, token=None):
    # 原有查询逻辑不变，仅给聚合字段加label
    current_user = info.context.user.id # 替换为你自己的当前咖啡馆ID获取逻辑
    cafe_customers = db.query(
        models.Customer.username, 
        func.count(models.Order.ordered_by).label("order_count"),
        func.sum(models.Order.price).label("total_spent")
    ).join(
        models.Order, models.Customer.id == models.Order.ordered_by
    ).filter(
        models.Order.cafe_id == current_user
    ).group_by(
        models.Customer.id
    ).all()

    # 关键逻辑：把查询返回的元组转成Graphene可识别的实例
    return [
        CafeCustomerStats(
            username=item.username,
            order_count=item.order_count,
            total_spent=float(item.total_spent)
        ) for item in cafe_customers
    ]

3. 调整GraphQL查询语句 #

现在可以按需查询需要的统计字段：

{
  customersByCafe(token: "") {
    username
    orderCount
    totalSpent
    avgSpentPerOrder
  }
}

返回结果会自动转为标准JSON格式，适配前端使用。

如果对金额精度要求高，可以把total_spent和avg_spent_per_order的类型改为graphene.Decimal，避免Float转换带来的精度丢失。

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