本文介绍如何通过 python 列表参数化构建安全、可扩展的 `create table` sql 语句，利用 `psycopg2.sql` 模块实现列名动态拼接，避免 sql 注入，支持任意数量的字段。
在实际数据库开发中，硬编码列名不仅难以维护，还容易引发 SQL 注入风险。psycopg2 提供了 sql.SQL 和 sql.Identifier 等安全构造工具，配合 Python 的字符串生成能力，可优雅地实现动态建表逻辑。
以下是一个健壮、可复用的 create_table 函数实现：
from psycopg2 import sql
def create_table(column_names: list[str], table_name: str = "my_table") -> sql.Composed:
"""
根据列名列表生成安全的 CREATE TABLE SQL 语句。
Args:
column_names: 字段名列表（如 ["col1", "col2", "colN"]）
table_name: 目标表名，默认为 "my_table"
Returns:
psycopg2.sql.Composed 对象，可直接用于 cursor.execute()
"""
# 安全拼接各列定义：每列类型为 BYTEA
column_defs = [sql.SQL("{} BYTEA").format(sql.Identifier(col)) for col in column_names]
# 构建完整 CREATE TABLE 语句（含固定字段 created_ts）
stmt = sql.SQL("""
CREATE TABLE {} (
created_ts TIMESTAMPTZ,
{}
)
""").format(
sql.Identifier(table_name),
sql.SQL(', ').join(column_defs)
)
return stmt✅ 关键优势说明：
- 使用 sql.Identifier() 自动转义列名，防止恶意输入（如 "col1; DROP TABLE users--"）导致注入；
- sql.SQL().join() 确保逗号分隔符与列定义之间无引号或空格错误；
- 返回 sql.Composed 类型，兼容 cursor.execute()，无需手动 .as_string(conn)；
- 支持自定义表名，提升复用性。
? 使用示例：
from psycopg2 import connect
conn = connect("dbname=test user=postgres")
cur = conn.cursor()
# 动态生成并执行建表语句
sql_stmt = create_table(["col1", "col2", "metadata"], table_name="uploads")
cur.execute(sql_stmt)
conn.commit()
cur.close()
conn.close()⚠️ 注意事项：
- 列名列表不能为空（否则会生成语法错误的 SQL），建议添加校验：
if not column_names: raise ValueError("At least one column name must be provided.") - 若需支持不同数据类型（如 col1 TEXT, col2 INTEGER），可将 column_names 替换为 columns: list[tuple[str, str]]（如 [("col1", "TEXT"), ("col2", "INTEGER")]），并相应调整 column_defs 构造逻辑；
- 生产环境建议配合事务与异常处理（try/except + conn.rollback()），确保建表失败时状态可控。
通过该方法，你既能保持代码简洁性，又能兼顾安全性与灵活性——这才是现代 Python 数据库开发的最佳实践。
