Psycopg3动态SQL：使用Identifier带别名安全获取JSON字段feature数据的方法探讨

我需要从JSON字段中获取feature数据，同时保留结果列的名称，因此希望了解使用Psycopg3动态定义列别名的更优、更安全的方式。

当前我已实现如下解决方案：

连接与配置代码 #

# imports
import psycopg
from psycopg import Connection, sql
from psycopg.rows import dict_row

# constants
project = "project_1"
location = "location_1"
data_table = "table_1"
features = ["feature_1", "feature_2"]
start_dt = "2024-04-22T16:00:00"
end_dt = "2024-04-22T17:00:00"

__user = "My"
__password = "I am supposed to be extra-complicated!"
__host = "localhost"
__database = "db"
__port = 5432

# connection
connection = psycopg.connect(
    user=__user,
    password=__password,
    host=__host,
    dbname=__database,
    port=__port,
    row_factory=dict_row,
)

查询构造代码 #

# Adapted from https://github.com/psycopg/psycopg2/issues/791#issuecomment-429459212
def alias_identifier(
    ident: str | tuple[str], alias: str | None = None
) -> sql.Composed:
    """Return a SQL identifier with an optional alias."""
    if isinstance(ident, str):
        ident = (ident,)
    if not alias:
        return sql.Identifier(*ident)
    # fmt: off
    return sql.Composed([sql.Literal(*ident), sql.SQL(" AS "),
                         sql.Identifier(alias)])
    # fmt: on

# source query str
QUERY = """SELECT
    current_database() AS project,
    timestamp,
    location,
    feature -> {feature}
FROM {data_table}
WHERE lower(location) = {location}
AND timestamp BETWEEN {start_dt} AND {end_dt}
"""

# SQL query
query = sql.SQL(QUERY).format(
    feature=sql.SQL(", feature -> ").join([alias_identifier(m, alias=m) for m in features]),
    data_table=sql.Identifier(data_table),
    location=sql.Literal(location),
    start_dt=sql.Literal(start_dt),
    end_dt=sql.Literal(end_dt),
)


print(query.as_string(connection))

执行输出SQL #

SELECT
    current_database() AS project,
    timestamp,
    location,
    feature -> 'feature_1' AS "feature_1", feature -> 'feature_2' AS "feature_2"
FROM "table_1"
WHERE lower(location) = 'location_1'
AND timestamp BETWEEN '2024-04-22T16:00:00' AND '2024-04-22T17:00:00'

该方案能得到预期结果，但我不确定是否违反Psycopg的使用规范，也想知道是否存在更优的实现方法。

解答 #

1. 当前方案的安全性验证 #

你的实现没有违反Psycopg的使用规范，核心逻辑是通过sql.Identifier处理数据库对象名（如表名、列别名），用sql.Literal处理值类型参数，完全规避了SQL注入风险，这是正确的安全实践方向。

2. 优化后的实现方案 #

针对你的场景，推荐以下更贴合Psycopg最佳实践的优化方式：

优化点说明

• 用参数绑定替代sql.Literal处理值参数：这是Psycopg官方推荐的方式，不仅更安全，还能让数据库缓存查询计划，提升重复执行的性能。
• 简化列构造逻辑：针对一级JSON键的提取场景，简化动态列生成函数，提升代码可读性。
• 分离查询模板与动态参数：让代码结构更清晰，维护性更强。

优化后完整代码

import psycopg
from psycopg import sql
from psycopg.rows import dict_row

# 配置参数
project = "project_1"
location = "location_1"
data_table = "table_1"
features = ["feature_1", "feature_2"]
start_dt = "2024-04-22T16:00:00"
end_dt = "2024-04-22T17:00:00"

# 数据库连接
connection = psycopg.connect(
    user="My",
    password="I am supposed to be extra-complicated!",
    host="localhost",
    dbname="db",
    port=5432,
    row_factory=dict_row,
)

def build_feature_column(feature_name: str) -> sql.Composed:
    """构造带别名的JSON字段提取列"""
    return sql.Composed([
        sql.SQL("feature -> {}").format(sql.Literal(feature_name)),
        sql.SQL(" AS "),
        sql.Identifier(feature_name)
    ])

# 基础查询模板（用%s作为值参数占位符）
QUERY_TEMPLATE = """SELECT
    current_database() AS project,
    timestamp,
    location,
    {features}
FROM {data_table}
WHERE lower(location) = %s
AND timestamp BETWEEN %s AND %s
"""

# 动态组合查询语句
query = sql.SQL(QUERY_TEMPLATE).format(
    features=sql.SQL(", ").join(build_feature_column(f) for f in features),
    data_table=sql.Identifier(data_table)
)

# 执行查询（通过参数绑定传递值）
with connection.cursor() as cur:
    cur.execute(query, (location, start_dt, end_dt))
    results = cur.fetchall()

# 调试用：打印生成的SQL语句
print(query.as_string(connection))

3. 额外注意事项 #

• 如果需要处理嵌套JSON键（如feature -> 'level1' -> 'level2'），可以修改build_feature_column函数，通过循环拼接JSON操作符：

  def build_nested_feature_column(keys: list[str]) -> sql.Composed:
      parts = [sql.SQL("feature")]
      for key in keys:
          parts.append(sql.SQL(" -> {}").format(sql.Literal(key)))
      column_expr = sql.Composed(parts)
      return sql.Composed([column_expr, sql.SQL(" AS "), sql.Identifier(keys[-1])])
  

• 始终使用sql.Identifier处理数据库对象名（表、列、别名等），避免直接拼接字符串，防止SQL注入。

内容的提问来源于stack exchange，提问作者Pietro D'Antuono