在 PostgreSQL 里跟 JSON 字段打交道时，经常需要按 key 动态提取多个字段，同时保留原始 key 名作为结果列的别名——比如 feature -> 'feature_1' AS "feature_1"。这事儿听起来简单，但真正的难点在于：既要防 SQL 注入，又要支持运行时动态列名。而 psycopg3 里sql.Identifier和sql.Literal的职责分界，正是解决这个问题的关键。
在实际项目中，PostgreSQL 的 JSON 字段（比如一张表里有个 feature 列，存的是一堆 JSON 对象）经常需要按需提取。举个例子，你可能想动态地取出 feature_1、feature_2 这些键，并且在查询结果里直接以 "feature_1" 作为列名。这时候，如果直接拼字符串写 SQL，一不留神就埋下了注入漏洞；如果全部用参数绑定，别名又没法动态生成。那么，正确的姿势到底是什么？
✅ 正确做法：分离「结构」与「参数」，用 format() 固定结构，execute() 绑定值
psycopg3 的最佳实践里有一条核心原则：标识符（表名、列名、别名）必须通过 sql.Identifier 动态构造；而用户输入的值（比如 JSON 键名、时间范围、位置等）统一走参数化查询，用 %(name)s 占位符加字典参数传入。这俩绝对不能混着用。
你可能会问：那我直接在代码里把 feature 列表拼进 feature -> {feature}，再用 alias_identifier 构造 AS "xxx"，行不行？功能上确实能跑，但有两个隐患：
alias_identifier里对alias用sql.Identifier(alias)是对的，但前提是alias必须来自可信输入（比如你写死的固定列表）。如果它来自 Web 表单之类的外部输入，一定得先校验它是不是合法标识符——只能包含字母、数字、下划线，还不能以数字开头。- 更关键的是：
feature -> 'feature_1'里的'feature_1'本质上是一个 JSON 键的字符串字面量，属于 运行时值，应该用%(feature_key)s参数化绑定，而不是硬拼进 SQL 结构里。否则不仅没法复用预编译语句，还容易出 bug。
下面给出一个推荐的重构方案，把结构部分和参数部分彻底分开：
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 的查询计划缓存，是生产环境最稳妥的做法。
