如何在SQLAlchemy的SELECT查询中将表行聚合为列表？

要实现将Table3的关联行聚合为JSON列表，与Table1、Table2的列一同返回，核心是利用SQLAlchemy的func调用对应数据库的聚合函数，将Table3的行转换为JSON对象后再聚合。以下分两种常见数据库场景实现：

1. PostgreSQL 场景 #

PostgreSQL支持json_build_object构造单行的JSON对象，配合array_agg聚合为列表：

from sqlalchemy import func, select

# 构造Table3的JSON对象聚合列
table3_agg = func.array_agg(
    func.json_build_object(
        "table3_col_a", Table3.table3_col_a,
        "table3_col_b", Table3.table3_col_b
    )
).label("table3_col_agg")

# 完整查询
query = (
    select(Table1, Table2, table3_agg)
    .join(Table2, Table2.id == Table1.id, isouter=True)
    .join(Table3, Table3.id == Table1.id, isouter=True)
    .group_by(Table1.id, Table2.id)
    .where(Table1.id == "id_1")
)

注：SQLAlchemy中full=True已废弃，推荐使用外连接语法保证即使Table2/Table3无关联行也能返回结果。

2. MySQL 场景 #

MySQL使用json_object构造单行JSON，json_arrayagg聚合为列表：

from sqlalchemy import func, select

table3_agg = func.json_arrayagg(
    func.json_object(
        "table3_col_a", Table3.table3_col_a,
        "table3_col_b", Table3.table3_col_b
    )
).label("table3_col_agg")

query = (
    select(Table1, Table2, table3_agg)
    .fullouterjoin(Table2, Table2.id == Table1.id)
    .fullouterjoin(Table3, Table3.id == Table1.id)
    .group_by(Table1.id, Table2.id)
    .where(Table1.id == "id_1")
)

结果验证 #

执行上述查询后，返回的单行结果中table3_col_agg字段会是符合期望的JSON列表：

[{"table3_col_a":"table3_col_a_cont1","table3_col_b":"table3_col_b_cont1"},{"table3_col_a":"table3_col_a_cont2","table3_col_b":"table3_col_b_cont2"}]

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