如何通过SQLAlchemy ORM对Int字段与Array[int]字段进行分组聚合？

核心思路是利用PostgreSQL的unnest函数拆分数组字段，再通过UNION ALL将first_col的单个值与拆分后的数组元素合并为统一数据集，最后在SQL层面完成分组统计，完全替代Python端的循环计数逻辑。

代码实现 #

首先补充导入func（SQLAlchemy的函数工具）：

from sqlalchemy import func, Column

然后构建聚合查询：

def get_some_stuff_and_aggregate() -> dict:
    # 子查询1：提取first_col作为独立的id项
    subq_first = select(some_table.c.first_col.label("some_id")).select_from(some_table)
    # 过滤掉first_col为NULL的情况（可选，根据业务需求调整）
    subq_first = subq_first.filter(some_table.c.first_col.is_not(None))

    # 子查询2：拆分second_col数组，将每个元素转为独立行
    subq_second = select(
        func.unnest(some_table.c.second_col).label("some_id")
    ).select_from(some_table)
    # 过滤掉空数组或NULL的情况（可选，根据业务需求调整）
    subq_second = subq_second.filter(
        some_table.c.second_col.is_not(None),
        func.array_length(some_table.c.second_col, 1) > 0
    )

    # 合并两个子查询的结果
    combined_subq = subq_first.union_all(subq_second)

    # 分组统计每个id出现的次数
    query = select(
        combined_subq.c.some_id,
        func.count(combined_subq.c.some_id).label("count")
    ).group_by(combined_subq.c.some_id)

    # 执行查询并转换为字典
    query_res = ...  # 执行查询的逻辑（如conn.execute(query).fetchall()）
    return {row.some_id: row.count for row in query_res}

关键逻辑说明 #

1. unnest函数：PostgreSQL专属函数，用于将数组字段拆分为多行，每个数组元素对应一行记录。
2. UNION ALL：将first_col的单行数据与拆分后的数组行合并，保证所有需要统计的id都在同一个数据集中。
3. 分组统计：通过GROUP BY some_id和COUNT()直接在数据库层面完成计数，避免了Python端循环遍历的性能损耗。

可选优化 #

• 如果业务允许忽略NULL值或空数组，保留代码中的过滤条件可以减少无效数据的处理。
• 若需要对统计结果排序，可以在主查询末尾添加.order_by(func.count(combined_subq.c.some_id).desc())。

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