如何在SQLAlchemy中为jsonb_agg内的jsonb_build_object添加ORDER BY？

针对你需要在SQLAlchemy的CTE里实现PostgreSQL原生SQL中jsonb_agg(...) ORDER BY pdb.score(notes.id) DESC的逻辑，这里提供两种可行方案：

方案1：用SQLAlchemy func调用自定义函数（ORM风格） #

如果已经在SQLAlchemy中注册了pdb.score这个自定义函数，直接通过func调用并指定排序：

from sqlalchemy import func, select
from your_module import Note  # 替换为你的Notes表模型

# 构建带排序的聚合CTE
notes_cte = select(
    # 按需要分组的字段，比如note所属的父ID
    Note.parent_id,
    func.jsonb_agg(
        # 聚合生成jsonb的字段，按需调整
        func.jsonb_build_object(
            "note_id", Note.id,
            "content", Note.content
        ).order_by(func.pdb.score(Note.id).desc())  # 核心：指定聚合内的排序逻辑
    ).label("sorted_notes")
).group_by(Note.parent_id).cte("notes_cte")

方案2：用text直接嵌入原生SQL片段（快速实现） #

如果不想额外注册自定义函数，直接用text把原生排序逻辑嵌入进去，和你的原始SQL完全对齐：

from sqlalchemy import func, select, text
from your_module import Note

notes_cte = select(
    Note.parent_id,
    func.jsonb_agg(
        func.jsonb_build_object(
            "note_id", Note.id,
            "content", Note.content
        ).order_by(text("pdb.score(notes.id) DESC"))  # 直接使用原生排序语句
    ).label("sorted_notes")
).group_by(Note.parent_id).cte("notes_cte")

关键说明 #

• 两种方案最终生成的SQL都会包含jsonb_agg(...) ORDER BY pdb.score(notes.id) DESC的逻辑，和你原有的可运行原生SQL一致。
• 方案1更贴合SQLAlchemy的ORM设计，适合长期维护的项目；方案2更灵活，适合快速迁移原生SQL到SQLAlchemy的场景。

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