SQLAlchemy 如何将每条记录的 JSON 键合并为数组？

2026-11-04数据阿明

可以在相关子查询的 FROM 中展开 JSON 的键，再用 array_agg() 汇总。这样外层查询对每条记录只返回一行，也避免了将集合返回函数直接放入聚合函数时报错。将结果列命名为 json_keys，还能避免与 Row.keys 方法冲突。

为什么原查询会拆成多行？

json_object_keys() 和 jsonb_object_keys() 返回的是 setof text，每个顶层键对应一行。将它们直接放在查询列中，同一条记录的 id 就会随各个键重复出现。PostgreSQL JSON 函数文档

原示例中的两个现象原因不同：

同一个 id 返回多行，是因为集合返回函数展开了键。
row.keys 显示 <bound method Row.keys ...>，是因为结果列名 keys 与当前 Row 对象的方法重名。日志中的 (1, 'key1') 表明键值已经返回，只是通过属性访问时取到了方法。

随后尝试的 array_agg(jsonb_object_keys(...)) 将集合返回函数直接放入聚合函数的参数中，因而报错。添加 GROUP BY 无法消除这种嵌套。

SQLAlchemy 1.4.52 的查询写法

下面的代码按模型声明的 postgresql.JSON 类型使用 json_object_keys()。代码假定示例记录已经写入数据库，或已经加入当前 Session 并完成必要的 flush。

import model
from sqlalchemy import Text, cast, column, func, select
from sqlalchemy.dialects.postgresql import (
    ARRAY,
    aggregate_order_by,
    array,
)

MyModel = model.MyModel

# 将集合返回函数放在子查询的 FROM 中。
key_rows = (
    func.json_object_keys(MyModel.extra_data)
    .table_valued(column("key", Text))
    .render_derived(name="jkeys")
)

# 为外层的当前记录聚合所有键。
keys_subquery = (
    select(
        func.array_agg(
            aggregate_order_by(key_rows.c.key, key_rows.c.key)
        )
    )
    .select_from(key_rows)
    .correlate(MyModel)
    .scalar_subquery()
)

results = (
    meta.session.query(
        MyModel.id,
        func.coalesce(
            keys_subquery,
            cast(array([]), ARRAY(Text)),
        ).label("json_keys"),
    )
    .order_by(MyModel.id)
    .all()
)

print("Result rows:")
for row in results:
    print(f"Row ID: {row.id}, Keys: {row.json_keys}")

table_valued() 将函数结果作为表使用，render_derived() 生成带列名的别名，例如 AS jkeys(key)；aggregate_order_by() 指定数组内键的顺序。参见 SQLAlchemy 1.4 PostgreSQL 文档。

.scalar_subquery() 将聚合查询作为外层查询的一列，.correlate(MyModel) 明确让子查询引用外层的当前记录。外层不需要再写 GROUP BY。SQLAlchemy 相关子查询文档

题目中两条记录的结果如下。Python 列表打印时，元素之间会有逗号：

Result rows:
Row ID: 1, Keys: ['key1', 'key2']
Row ID: 2, Keys: ['key3', 'key4']

为什么这能解决聚合报错？

上述查询的 SQL 结构可以理解为下面这样。my_model 仅为示意表名，实际应以模型映射的表名为准：

SELECT
    m.id,
    COALESCE(
        (
            SELECT array_agg(jkeys.key ORDER BY jkeys.key)
            FROM json_object_keys(m.extra_data) AS jkeys(key)
        ),
        ARRAY[]::text[]
    ) AS json_keys
FROM my_model AS m
ORDER BY m.id;

json_object_keys() 先在子查询的 FROM 中产生键行，array_agg() 再聚合这些行中的普通列，避免了 array_agg(json_object_keys(...)) 的直接嵌套。

错误提示中的 LATERAL 是另一种组织查询的方法。本例通过相关标量子查询处理每条记录，无需显式写出外层 LATERAL JOIN。

类型、空对象和结果读取要注意什么？

模型示例声明的是 postgresql.JSON，查询却调用了 jsonb_object_keys()。如果实际数据库列也是 JSON，应使用上面代码中的 json_object_keys()；如果实际列是 JSONB，则改用 jsonb_object_keys()。模型声明与所述运行结果存在差异，不能仅凭片段判断数据库中的实际类型。

对于 {}，键函数不会产生行，array_agg() 因此返回 SQL NULL。代码中的 coalesce(..., CAST(ARRAY[] AS TEXT[])) 将其转换为空数组，经 psycopg2 读取后得到 []，同时保留这条外层记录。PostgreSQL 聚合函数文档

nullable=False 不能保证 JSON 内容是对象。数组、字符串或 JSON null 需要按业务规则另行处理。这里仅提取顶层键，不会递归遍历嵌套对象。

结果列名应避免使用 keys。如果必须保留 .label("keys")，可以通过映射访问取值：

for row in results:
    print(row._mapping["keys"])

这种读取方式只适用于列标签确实为 keys 的查询；主示例已经改成 json_keys，直接使用 row.json_keys 即可。