SQLAlchemy 1.4中PostgreSQL子查询结合json_agg报错求助

需求与原生SQL #

需要实现关联table_a与user表，同时通过子查询结合json_agg获取符合条件的items数据，对应的原生SQL如下：

SELECT 
    "table_a".id,
    "table_a".body,
    "table_a".created_at,
    "table_a".archived_at,
    "user".full_name,
    (
        SELECT json_agg(items_alias_b)
        FROM (
            SELECT items_alias_a.id, items_alias_a.type, items_alias_a.filename, items_alias_a.s3_original_key
            FROM items items_alias_a
            WHERE items_alias_a.status = 'completed'::itemsstatusenum
                AND items_alias_a.id = ANY("table_a".item_ids)
        ) items_alias_b
    ) AS "items"
FROM "table_a"
JOIN "user" ON "user".id = "table_a".created_by
WHERE "table_a".other_id = 'XXXXXXXX'
ORDER BY "table_a".id DESC

原代码问题与错误日志 #

原代码运行时报错(psycopg2.errors.SyntaxError) subquery must return only one column，生成的错误SQL如下：

SELECT 
  table_a.id AS table_a_id, 
  table_a.body AS table_a_body, 
  table_a.created_at AS table_a_created_at, 
  table_a.archived_at AS table_a_archived_at, 
  "user".full_name AS full_name, 
  (
    SELECT 
      json_agg(
        (
          SELECT 
            item.id, 
            item.type, 
            item.name
          FROM 
            item, 
            table_a 
          WHERE 
            item.status = %(status_1)s 
            AND item.id = any(table_a.item_ids)
        )
      ) AS json_agg_1
  ) AS item 
FROM 
  table_a 
  JOIN "user" ON "user".id = table_a.created_by 
WHERE 
  table_a.other_id = %(other_id_1)s
ORDER BY 
  table_a.id DESC

原代码核心问题：

• 直接将多列子查询传入json_agg，导致生成的SQL在json_agg内部嵌套了返回多列的子查询，触发语法错误。
• 子查询未正确关联外层的table_a，错误地将table_a加入子查询的FROM列表，而非引用当前行的item_ids。

修正后的代码 #

from sqlalchemy import func, select

# 构造内层子查询：获取符合条件的item字段（与原生SQL的内层SELECT对应）
items_inner_subq = select(
    Model.Item.id,
    Model.Item.type,
    # 注意：原生SQL中是filename和s3_original_key，需根据实际Model调整字段
    Model.Item.filename,
    Model.Item.s3_original_key
).filter(
    Model.Item.status == Model.ItemStatusEnum.completed,
    # 关联外层table_a的当前行item_ids
    Model.Item.id == func.any(Model.TableA.item_ids)
).subquery()

# 构造最终查询
query = db_session.query(
    Model.TableA.id,
    Model.TableA.body,
    Model.TableA.created_at,
    Model.TableA.archived_at,
    Model.User.full_name.label("full_name"),
    # 用scalar_subquery实现原生SQL中的嵌套子查询结构
    select(func.json_agg(items_inner_subq))
    .scalar_subquery()
    .label("items")
).join(
    Model.User,
    Model.User.id == Model.TableA.created_by,
).filter(
    Model.TableA.other_id == other_id,
).order_by(Model.TableA.id.desc())

messages = query.all()

关键修正点 #

1. 子查询结构匹配：通过select(func.json_agg(items_inner_subq)).scalar_subquery()完全复刻原生SQL中(SELECT json_agg(...) FROM (...)) AS items的嵌套结构，避免json_agg内部嵌套多列子查询。
2. 正确关联外层表：内层子查询通过func.any(Model.TableA.item_ids)引用外层table_a的当前行数据，不会错误地将table_a加入子查询的FROM列表。
3. 字段对齐：确保内层子查询的字段与原生SQL一致（原Model中缺少filename和s3_original_key，需补充后替换代码中的对应字段）。

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