如何将DISTINCT SQL查询结果按运单分组为嵌套对象？

一、用SQL直接生成嵌套结构 #

如果你的数据库支持JSON聚合函数，可直接在SQL层面完成嵌套结构生成，省去后续Python处理步骤。以下是主流数据库的实现示例：

PostgreSQL 写法 #

SELECT
    sd.erp_order AS shipment_id,
    json_agg(
        json_build_object(
            'product_code', th.product_code,
            'lot', th.lot
        )
    ) AS products
FROM transaction_history th
JOIN shipment_detail sd
    ON sd.shipment_id = th.reference_id
    AND sd.item = th.item
WHERE th.transaction_type = '1'
    AND sd.erp_order IN ('1111', '1112')
GROUP BY sd.erp_order;

MySQL 8.0+ 写法 #

SELECT
    sd.erp_order AS shipment_id,
    JSON_ARRAYAGG(
        JSON_OBJECT(
            'product_code', th.product_code,
            'lot', th.lot
        )
    ) AS products
FROM transaction_history th
JOIN shipment_detail sd
    ON sd.shipment_id = th.reference_id
    AND sd.item = th.item
WHERE th.transaction_type = '1'
    AND sd.erp_order IN ('1111', '1112')
GROUP BY sd.erp_order;

执行后会直接返回包含嵌套数组的JSON结构，可直接用于Django API输出。

二、用Python/Django处理扁平结果 #

若数据库不支持JSON聚合，或更倾向于在应用层处理，可通过Python代码将扁平查询结果转换为嵌套结构：

方法1：用itertools.groupby分组处理 #

先执行原有SQL获取扁平数据，再通过分组生成嵌套结构：

from itertools import groupby
from django.db import connection

def get_shipment_products():
    with connection.cursor() as cursor:
        cursor.execute("""
            SELECT DISTINCT
                sd.erp_order AS shipment_id, 
                th.product_code, 
                th.lot
            FROM transaction_history th
            JOIN shipment_detail sd
                ON sd.shipment_id = th.reference_id
                AND sd.item = th.item
            WHERE th.transaction_type = '1'
                AND sd.erp_order in ('1111', '1112')
        """)
        rows = cursor.fetchall()
        
        result = []
        # 按运单ID分组
        for shipment_id, group in groupby(rows, key=lambda x: x[0]):
            products = []
            for item in group:
                products.append({
                    "product_code": item[1],
                    "lot": item[2]
                })
            result.append({
                "shipment_id": shipment_id,
                "products": products
            })
        return result

方法2：用Django序列化器（已有模型时） #

若项目中已定义对应Django模型，可通过预关联查询+序列化器实现：

from django.db.models import Prefetch
from rest_framework import serializers
from .models import ShipmentDetail, TransactionHistory

class ProductSerializer(serializers.Serializer):
    product_code = serializers.CharField()
    lot = serializers.CharField()

class ShipmentSerializer(serializers.Serializer):
    shipment_id = serializers.CharField(source='erp_order')
    products = ProductSerializer(many=True)

def get_shipment_data():
    # 预筛选符合条件的交易记录
    filtered_transactions = TransactionHistory.objects.filter(transaction_type='1')
    # 查询运单并预关联目标交易记录
    shipments = ShipmentDetail.objects.filter(
        erp_order__in=['1111', '1112']
    ).prefetch_related(
        Prefetch(
            'transaction_history_set',
            queryset=filtered_transactions,
            to_attr='filtered_products'
        )
    )
    
    shipment_list = []
    for shipment in shipments:
        products = [
            {"product_code": th.product_code, "lot": th.lot}
            for th in shipment.filtered_products
        ]
        shipment_list.append({
            "shipment_id": shipment.erp_order,
            "products": products
        })
    return shipment_list

方案选择建议 #

• 数据库支持JSON聚合函数时，优先用SQL方案，减少应用层逻辑，性能更优。
• 数据库不支持或需要灵活业务处理时，选择Python/Django方案。

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