FastAPI中如何通过Join将同ID多值合并为单个列表

问题场景 #

执行Join查询后，同一product_id的记录重复返回，仅name字段不同。

当前输出 #

[{"product_id": 1,"model": "model_1","image": "catalog/product/blabla.jpg","price": 5800,"quantity": 999,"name": "name_1"},{"product_id": 1,"model": "model_1","image": "catalog/product/blabla.jpg","price": 5800,"quantity": 999,"name": "name_2"},{"product_id": 1,"model": "model_1","image": "catalog/product/blabla.jpg","price": 5800,"quantity": 999,"name": "name_3"}]

期望输出 #

[{"product_id": 1,"model": "model_1","image": "catalog/product/blabla.jpg","price": 5800,"quantity": 999,"name": ["name_1", "name_2", "name_3"]}]

解决方案 #

提供两种实现方式，按需选择：

方式一：Python层面聚合结果（通用兼容） #

修改orm.py中的查询函数，先获取所有记录，再聚合name字段：

def get_product_by_id(product_id: int, db: Session):
    # 查询所有关联记录，给重复字段加别名避免冲突
    results = db.query(
        OcProduct.product_id, 
        OcProduct.model, 
        OcProduct.image, 
        OcProduct.price, 
        OcProduct.quantity, 
        OcStockStatu.name.label('stock_name'),
        OcOptionValueDescription.name.label('option_name')
    ).join(
        OcStockStatu, OcProduct.stock_status_id == OcStockStatu.stock_status_id
    ).join(
        OcProductOptionValue, OcProductOptionValue.product_id == product_id
    ).join(
        OcOptionValueDescription, OcOptionValueDescription.option_value_id == OcProductOptionValue.option_value_id
    ).filter(OcProduct.product_id == product_id).all()
    
    if not results:
        return None
    
    # 聚合数据，提取所有option的name组成列表
    aggregated_product = {
        "product_id": results[0].product_id,
        "model": results[0].model,
        "image": results[0].image,
        "price": results[0].price,
        "quantity": results[0].quantity,
        "name": [r.option_name for r in results]
    }
    
    return [aggregated_product]

修改product.py接口，增加空值判断：

from fastapi import HTTPException

@router.get("/{product_id}")
def get_detail_product(product_id: int, db: Session = Depends(get_db)):
    product = orm_product.get_product_by_id(product_id, db)
    if not product:
        raise HTTPException(status_code=404, detail="商品不存在")
    return product

方式二：数据库层面聚合（性能更优） #

利用数据库聚合函数直接在查询时合并name字段，注意不同数据库函数不同：

PostgreSQL版本

from sqlalchemy import func

def get_product_by_id(product_id: int, db: Session):
    result = db.query(
        OcProduct.product_id, 
        OcProduct.model, 
        OcProduct.image, 
        OcProduct.price, 
        OcProduct.quantity,
        func.array_agg(OcOptionValueDescription.name).label('name')
    ).join(
        OcStockStatu, OcProduct.stock_status_id == OcStockStatu.stock_status_id
    ).join(
        OcProductOptionValue, OcProductOptionValue.product_id == product_id
    ).join(
        OcOptionValueDescription, OcOptionValueDescription.option_value_id == OcProductOptionValue.option_value_id
    ).filter(OcProduct.product_id == product_id).group_by(
        OcProduct.product_id, OcProduct.model, OcProduct.image, OcProduct.price, OcProduct.quantity
    ).first()
    
    if not result:
        return None
    
    aggregated_product = {
        "product_id": result.product_id,
        "model": result.model,
        "image": result.image,
        "price": result.price,
        "quantity": result.quantity,
        "name": result.name
    }
    
    return [aggregated_product]

MySQL版本

from sqlalchemy import func

def get_product_by_id(product_id: int, db: Session):
    result = db.query(
        OcProduct.product_id, 
        OcProduct.model, 
        OcProduct.image, 
        OcProduct.price, 
        OcProduct.quantity,
        func.group_concat(OcOptionValueDescription.name).label('name')
    ).join(
        OcStockStatu, OcProduct.stock_status_id == OcStockStatu.stock_status_id
    ).join(
        OcProductOptionValue, OcProductOptionValue.product_id == product_id
    ).join(
        OcOptionValueDescription, OcOptionValueDescription.option_value_id == OcProductOptionValue.option_value_id
    ).filter(OcProduct.product_id == product_id).group_by(
        OcProduct.product_id, OcProduct.model, OcProduct.image, OcProduct.price, OcProduct.quantity
    ).first()
    
    if not result:
        return None
    
    aggregated_product = {
        "product_id": result.product_id,
        "model": result.model,
        "image": result.image,
        "price": result.price,
        "quantity": result.quantity,
        "name": result.name.split(',') if result.name else []
    }
    
    return [aggregated_product]

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