Python/FastAPI+SQLAlchemy+PostgreSQL嵌套JSONB查询报错及优化方案

SQLAlchemy的text()方法会把语句里的$当成参数占位符（比如$1这种参数编号），但PostgreSQL的JSON路径表达式里的$是表示JSON根节点的语法，两者冲突，导致数据库解析时认为$后面缺少参数编号，直接抛出syntax error at or near $。

方法1：把JSON路径作为参数传递 #

避免SQLAlchemy解析路径里的$，直接把路径字符串作为参数传给SQL：

from sqlalchemy import text

target_key = "要查找的键名"
# 用:path作为SQLAlchemy的参数占位符，路径里的$由PostgreSQL解析
query = text("""
    SELECT jsonb_path_query_array(record_map, :path) AS matched_objects
    FROM private_notion
""")
# 路径表达式里的{{target_key}}是Python字符串格式化，生成带引号的键名
result = db.session.execute(query, {"path": f'$.."{{{target_key}}}"'}).fetchall()

方法2：用双美元符号包裹JSON路径 #

PostgreSQL支持用$$作为字符串分隔符，这样里面的$不会被SQLAlchemy识别为参数占位符：

query = text("""
    SELECT jsonb_path_query_array(record_map, $$ $.."要查找的键名" $$) AS matched_objects
    FROM private_notion
""")
result = db.session.execute(query).fetchall()

方案A：用jsonb_path（PostgreSQL 12+推荐） #

如果你的PostgreSQL版本是12及以上，jsonb_path_query_array是最简洁的方式，路径$.."target_key"会递归遍历所有层级，返回所有匹配key的对象数组。

方案B：递归CTE遍历（兼容旧版本） #

如果用的是PostgreSQL 11及以下，或者需要更灵活的处理逻辑，可以用递归CTE遍历所有嵌套对象：

from sqlalchemy import text

target_key = "要查找的键名"
query = text("""
    WITH RECURSIVE nested_objs AS (
        SELECT record_map AS obj
        FROM private_notion
        UNION ALL
        SELECT value AS obj
        FROM nested_objs, jsonb_each(obj)
        WHERE jsonb_typeof(obj) = 'object'
    )
    SELECT obj -> :target_key AS matched_obj
    FROM nested_objs
    WHERE obj ? :target_key
""")
result = db.session.execute(query, {"target_key": target_key}).fetchall()

这个方法会逐层拆解JSON对象，直到找到所有包含目标key的对象，适合复杂嵌套场景。

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