Django实现JsonField数组值按wh与created_at分组求和

方案1：PostgreSQL 原生SQL实现（大数据量首选） #

嵌套JSON数组的聚合无法直接通过基础ORM的annotate完成，核心逻辑是先通过数据库JSON函数把两层嵌套数组逐行拆平，再按日期和wh编号分组求和。PostgreSQL作为Django JsonField原生适配的数据库，用jsonb_array_elements拆数组性能最高。

对应SQL逻辑：

WITH flattened_stocks AS (
    SELECT
        DATE(sh.created_at) AS record_date,
        (stocks_item ->> 'wh')::INTEGER AS wh_id,
        (stocks_item ->> 'qty')::INTEGER AS qty
    FROM stocks_history sh
    -- 展开第一层wh_data数组
    CROSS JOIN jsonb_array_elements(sh.wh_data) AS wh_data_item
    -- 展开第二层每个条目下的stocks数组
    CROSS JOIN jsonb_array_elements(wh_data_item -> 'stocks') AS stocks_item
)
SELECT
    wh_id AS wh,
    SUM(qty) AS qty,
    record_date AS created_at
FROM flattened_stocks
GROUP BY record_date, wh_id
ORDER BY record_date, wh_id;

在Django中直接用raw方法调用即可：

# 执行原生查询
raw_results = StocksHistory.objects.raw(
    """
    WITH flattened_stocks AS (
        SELECT
            sh.id,
            DATE(sh.created_at) AS record_date,
            (stocks_item ->> 'wh')::INTEGER AS wh_id,
            (stocks_item ->> 'qty')::INTEGER AS qty
        FROM stocks_history sh
        CROSS JOIN jsonb_array_elements(sh.wh_data) AS wh_data_item
        CROSS JOIN jsonb_array_elements(wh_data_item -> 'stocks') AS stocks_item
    )
    SELECT
        wh_id AS wh,
        SUM(qty) AS qty,
        record_date AS created_at
    FROM flattened_stocks
    GROUP BY record_date, wh_id
    ORDER BY record_date, wh_id
    """
)

# 转换为要求的JSON结构
final_results = [
    {
        "wh": item.wh,
        "qty": item.qty,
        "created_at": item.created_at.isoformat()
    }
    for item in raw_results
]

方案2：Python内存聚合（中小数据量首选，全数据库兼容） #

如果单表总记录数在万级以内，完全没必要写数据库绑定的复杂JSON语法，直接查询需要的字段到内存中聚合，代码逻辑简单易维护，不限制数据库类型（MySQL、SQLite都能用）。

from collections import defaultdict

# 只查询需要的字段，减少数据传输量
qs = StocksHistory.objects.values("wh_data", "created_at")
agg_map = defaultdict(int)

for record in qs:
    record_day = record["created_at"].date()
    # 遍历第一层wh_data数组
    for entry in record["wh_data"]:
        # 遍历第二层stocks数组累加qty
        for stock in entry.get("stocks", []):
            agg_key = (record_day, stock["wh"])
            agg_map[agg_key] += stock["qty"]

# 转换为要求的输出格式
final_results = []
for (day, wh_id), total_qty in agg_map.items():
    final_results.append({
        "wh": wh_id,
        "qty": total_qty,
        "created_at": day.isoformat()
    })

补充说明：

• 如果需要实现示例中某日期下对应wh无库存记录时返回qty=0的效果，只需提前拉取全量wh编号列表、全量有记录的日期列表，遍历两者的组合作为key，聚合结果中不存在的key默认赋值0即可。
• 如果使用MySQL 8.0+版本，可将方案1中的jsonb_array_elements替换为JSON_TABLE函数实现同等的数组拆包逻辑，低版本MySQL建议直接用方案2。

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