SQLAlchemy 1.4编写Postgres嵌套JSON子查询出现CardinalityViolation错误

错误原因 #

你遇到的CardinalityViolation错误核心原因有两个：

1. 你单独定义的properties、features子查询未添加和外层Hardinfra行的关联过滤，作为标量子查询执行时会返回所有hardinfra表的行结果，不符合标量子查询只能返回单行单列的要求。
2. 最终的主查询没有指定查询来源为Hardinfra表，导致外层查询缺失Hardinfra的上下文，子查询中写的Hardinfra.id关联逻辑完全失效。

正确实现代码 #

你之前写的responses和protections两个关联标量子查询是正确的，不需要额外定义properties、features子查询，直接在主查询中构造JSON结构即可：

# 保留原有正确的两个关联标量子查询
responses = (
    db.session.query(
        func.json_object_agg(Expert.etype, Expert_Response.response).label("responses")
    )
    .join(Expert_Response.exp)
    .filter(Expert_Response.infra_id == Hardinfra.id)
    .scalar_subquery()
)

protections = (
    db.session.query(
        func.json_object_agg(Protection.ptype, Infra_Protection.pscore).label("protections")
    )
    .join(Infra_Protection.prot)
    .filter(Infra_Protection.infra_id == Hardinfra.id)
    .scalar_subquery()
)

# 直接构造主查询即可
q = db.session.query(
    func.json_build_object(
        "type", "FeatureCollection",
        "features", func.json_agg(
            func.json_build_object(
                "type", "Feature",
                "geometry", Hardinfra.geom,
                "properties", func.json_build_object(
                    "id", Hardinfra.id,
                    "responses", responses,
                    "protections", protections
                )
            )
        )
    ).label("allinfra")
).select_from(Hardinfra)

# 执行获取结果
result = q.scalar()

补充说明 #

如果确实需要拆分properties部分的逻辑，也不能直接定义为无过滤的标量子查询，需要确保子查询和外层行的关联，完全可以直接在主查询中嵌套构造，和原生SQL的写法逻辑完全对应，不需要额外拆分子查询引入额外的关联问题。

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