本文介绍了如何在 PostgreSQL 数据库中，使用 SQLAlchemy 和 Python 查询包含深度嵌套对象的 JSONB 列。我们将探讨如何使用 jsonb_path_query 函数以及 JSONPath 表达式来高效地检索所需数据，并解决常见的语法错误。通过本文，你将掌握一种更灵活、强大的 JSONB 数据查询方法。
理解 JSONB 和 JSONPath
PostgreSQL 的 JSONB 数据类型允许你存储 JSON（JavaScript Object Notation）数据，并对其进行高效的查询。JSONPath 是一种查询 JSON 数据的语言，类似于 XPath 用于 XML 数据。
在处理嵌套的 JSONB 对象时，直接访问深层嵌套的数据可能比较困难。这时，jsonb_path_query 函数结合 JSONPath 表达式就显得非常强大。
使用 jsonb_path_query 查询嵌套对象
假设我们有一个名为 private_notion 的表，其中包含一个名为 record_map 的 JSONB 列，该列存储了嵌套的 JSON 对象。我们的目标是根据特定的键（例如 UUID）在 record_map 中查找对象。
以下是一个示例 JSON 结构：
{
"blocks": {
"7a9abf0d-a066-4466-a565-4e6d7a960a37": {
"name": "block1",
"value": 1,
"child": {
"7a9abf0d-a066-4466-a565-4e6d7a960a37": {
"name": "block2",
"value": 2,
"child": {
"7a9abf0d-a066-4466-a565-4e6d7a960a37": {
"name": "block3",
"value": 3
}
}
},
"7a9abf0d-a066-4466-a565-4e6d7a960a38": {
"name": "block4",
"value": 4,
"child": {
"7a9abf0d-a066-4466-4466-a565-4e6d7a960a39": {
"name": "block5",
"value": 5,
"child": {
"7a9abf0d-a066-4466-a565-4e6d7a960a40": {
"name": "block6",
"value": 6
}
}
}
}
}
}
}
}
}要查找包含特定 UUID 的对象，可以使用以下 SQL 查询：
SELECT jsonb_path_query(record_map,
'strict $.**?(@.keyvalue().key==$target_id)',
jsonb_build_object('target_id',
'7a9abf0d-a066-4466-a565-4e6d7a960a37'))
FROM private_notion
WHERE site_id = '45bf37be-ca0a-45eb-838b-015c7a89d47b';这个查询使用了 jsonb_path_query 函数，并传入了以下参数：
- record_map: 要查询的 JSONB 列。
- 'strict $.**?(@.keyvalue().key==$target_id)': JSONPath 表达式，用于递归搜索 JSON 对象，查找键等于 $target_id 的对象。strict 模式确保了表达式的严格匹配。
- jsonb_build_object('target_id', '7a9abf0d-a066-4466-a565-4e6d7a960a37'): 创建一个 JSON 对象，将 target_id 设置为要查找的 UUID。
在 SQLAlchemy 中使用 jsonb_path_query
在 SQLAlchemy 中，可以使用 text 方法执行原始 SQL 查询。以下是一个示例：
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession
async def get_private_notion_page(
site_uuid: str, page_id: str, db_session: AsyncSession
) -> dict:
"""
Retrieves a nested object from a JSONB column by key using jsonb_path_query.
"""
query = text(
"""
SELECT jsonb_path_query(record_map,
'strict $.**?(@.keyvalue().key==$target_id)',
jsonb_build_object('target_id', :page_id))
FROM private_notion
WHERE site_id = :site_uuid
"""
)
result = await db_session.execute(query, {"page_id": page_id, "site_uuid": site_uuid})
result = result.scalars().first()
return result在这个例子中，我们使用了参数化查询，将 page_id 和 site_uuid 作为参数传递给查询，避免了 SQL 注入的风险。
常见错误和解决方法
在尝试使用 jsonb_path_query 时，可能会遇到一些常见的错误。以下是一些解决方法：
- 语法错误： 确保 JSONPath 表达式使用单引号括起来。
- UUID 格式错误： 确保 UUID 在 JSONPath 表达式中用双引号括起来。
- 未启用 strict 模式： 建议在使用 .** 访问器时，始终启用 strict 模式，以避免意外的结果。
使用 SQLAlchemy JSONPath 类型
从 SQLAlchemy 2.0 开始，你可以使用 JSONPath 类型来更安全地传递 JSONPath 表达式。
from sqlalchemy.dialects.postgresql import JSONPath
from sqlalchemy import column, table, select
private_notion_table = table(
"private_notion",
column("record_map"),
column("site_id"),
)
def get_private_notion_page(site_uuid: str, page_id: str):
"""
Retrieves a nested object from a JSONB column by key using jsonb_path_query and SQLAlchemy JSONPath.
"""
target_id = "7a9abf0d-a066-4466-a565-4e6d7a960a37"
jsonpath_expression = "strict $.**?(@.keyvalue().key==$target_id)"
stmt = select(
func.jsonb_path_query(
private_notion_table.c.record_map,
jsonpath_expression,
func.jsonb_build_object("target_id", target_id),
)
).where(private_notion_table.c.site_id == site_uuid)
# Execute the statement using your database session
# result = await db_session.execute(stmt)
# return result.scalars().first()
return stmt # Returning the statement for demonstration总结
通过本文，你学习了如何使用 PostgreSQL 的 jsonb_path_query 函数和 JSONPath 表达式，结合 SQLAlchemy，高效地查询嵌套的 JSONB 数据。 掌握这些技术，可以让你更灵活地处理 JSONB 数据，并构建更强大的应用程序。记住，正确地使用 JSONPath 表达式，并注意常见的错误，是成功查询 JSONB 数据的关键。
