本文介绍在 psycopg3 中安全、高效地为 json 字段提取操作（如 feature -> 'key'）动态生成带语义化别名的 sql 查询，避免 sql 注入风险，同时兼顾可维护性与性能。
本文介绍在 psycopg3 中安全、高效地为 json 字段提取操作（如 feature -> 'key'）动态生成带语义化别名的 sql 查询，避免 sql 注入风险，同时兼顾可维护性与性能。
在处理 PostgreSQL 中的 JSON 字段（如 feature 列存储 JSON 对象）时，常需按键动态提取多个字段并保留其原始名称作为结果列别名（例如 feature -> 'feature_1' AS "feature_1"）。关键挑战在于：既要防止 SQL 注入，又要支持运行时动态列名——而这恰恰是 sql.Identifier 和 sql.Literal 的职责分界所在。
✅ 正确做法：分离「结构」与「参数」，用 format() 固定结构，execute() 绑定值
psycopg3 的最佳实践强调：标识符（表名、列名、别名）必须通过 sql.Identifier 动态构造；而用户输入的值（如 JSON 键名、时间范围、位置等）应统一通过参数化查询（%(name)s 占位符 + 字典参数）传入。二者不可混用。
你当前代码中将 feature 列表直接拼入 feature -> {feature} 并用 alias_identifier 构造 AS "xxx"，虽功能可行，但存在两个隐患：
Comprehensive Three.js 3D graphics reference
下载
详细的 Three.js 3D 图形参考，涵盖场景设置、相机、几何体、材质、光照、动画、控制器、加载器、数学工具和调试。
- alias_identifier 中对 alias 使用 sql.Identifier(alias) 是正确的，但若 alias 来自不可信输入（如 Web 表单），必须确保已校验其为合法标识符（仅含字母、数字、下划线，不以数字开头）；
- 更重要的是，feature -> 'feature_1' 中的 'feature_1' 实际是 JSON 键的字符串字面量，属于 运行时值，应使用 %(feature_key)s 参数化，而非拼入 SQL 结构——否则无法复用预编译语句，且易出错。
推荐重构如下：
from psycopg import sql, connect
from psycopg.rows import dict_row
# 安全的动态列生成函数（仅用于标识符/别名）
def safe_column_with_alias(json_key: str) -> sql.Composed:
# json_key 作为别名必须是合法标识符（建议提前校验）
if not json_key.isidentifier():
raise ValueError(f"Invalid alias name: {json_key!r}")
return sql.SQL("feature -> %(key)s AS ").compose(
sql.Identifier(json_key)
)
# 构建静态 SQL 模板：结构固定，仅留参数占位符
QUERY_TEMPLATE = sql.SQL("""
SELECT
current_database() AS project,
timestamp,
location,
{json_columns}
FROM {table}
WHERE lower(location) = %(location)s
AND timestamp BETWEEN %(start_dt)s AND %(end_dt)s
""")
# 动态生成所有 feature -> key AS "key" 子句
features = ["feature_1", "feature_2"]
json_columns = sql.SQL(", ").join(
safe_column_with_alias(f) for f in features
)
# 格式化结构部分（表名用 Identifier）
query = QUERY_TEMPLATE.format(
json_columns=json_columns,
table=sql.Identifier("table_1")
)
# 执行时传入所有运行时值（包括 JSON 键名！）
params = {
"location": "location_1",
"start_dt": "2024-04-22T16:00:00",
"end_dt": "2024-04-22T17:00:00",
}
# 注意：每个 feature 键需单独传参（因 psycopg 不支持列表参数展开）
# → 改用循环或拼接参数字典
for feat in features:
params[f"feat_{feat}"] = feat
# 若需单次查询提取多键，推荐改写为：feature -> %(feat_feature_1)s
# 但更简洁的方式是：预先生成完整参数字典
full_params = {**params}
for feat in features:
full_params[f"key_{feat}"] = feat
# 最终查询（示例中 features = ['feature_1','feature_2']）
# feature -> %(key_feature_1)s AS "feature_1", feature -> %(key_feature_2)s AS "feature_2"
dynamic_select = sql.SQL(", ").join(
sql.SQL("feature -> %(key_{})s AS ").format(**{f"key_{f}": sql.Identifier(f)}).compose(sql.Identifier(f))
for f in features
)
# ⚠️ 实际应用中建议用辅助函数封装此逻辑✅ 更简洁实用的方案（推荐）
对于多数场景，无需过度抽象，直接组合即可：
features = ["feature_1", "feature_2"]
# 1. 构建 SELECT 子句（安全使用 Identifier 生成别名）
select_parts = []
params = {"location": "location_1", "start_dt": "...", "end_dt": "..."}
for feat in features:
select_parts.append(
sql.SQL("feature -> %(key_{})s AS ").format(**{f"key_{feat}": sql.SQL(feat)}).compose(sql.Identifier(feat))
)
params[f"key_{feat}"] = feat # 键值本身作为参数值（注意：此处 feat 是字符串字面量，非用户输入）
select_clause = sql.SQL(", ").join(select_parts)
# 2. 组装完整查询
query = sql.SQL("""
SELECT current_database() AS project, timestamp, location, {select}
FROM {table}
WHERE lower(location) = %(location)s
AND timestamp BETWEEN %(start_dt)s AND %(end_dt)s
""").format(
select=select_clause,
table=sql.Identifier("table_1")
)
# 3. 执行
with connection.cursor(row_factory=dict_row) as cur:
cur.execute(query, params)
results = cur.fetchall()⚠️ 关键注意事项
- 永远不要用 str.format() 或 f-string 拼接 SQL 片段：极易引入 SQL 注入。
- sql.Identifier() 只接受字符串或元组，且内容必须为合法 SQL 标识符：若 feat 来自用户输入，请先验证 feat.isidentifier() 或使用白名单。
- JSON 键名（如 'feature_1'）属于数据值，不是标识符：应通过 %(param)s 参数化，而非 sql.Identifier。
- 别名（AS 后的名称）是标识符：必须用 sql.Identifier(alias) 包裹。
- 避免重复构建查询：对固定结构（如表名、字段模板）用 sql.SQL().format() 一次生成；对变化值（时间、位置、键名）统一走参数绑定。
遵循以上原则，既能保证安全性，又能获得 psycopg3 的查询计划缓存优势，是生产环境的最佳实践。
