如何优化PYODBC MERGE操作的Python代码安全性？防范SQL注入

你的当前代码已经具备SQL注入防护能力 #

你现在的写法是安全的参数化查询实现，不会存在SQL注入风险。原因在于：

• 你用?作为参数占位符，而非直接将数据拼接进SQL字符串
• 所有业务数据都通过execute的第二个参数传入，pyodbc驱动会自动处理参数的转义和类型转换，确保数据不会被解析为SQL指令的一部分

这种批量MERGE的写法是合理的，也是pyodbc中处理批量数据合并的常用方式。

进一步的安全性与代码质量优化建议 #

1. 使用上下文管理器自动管理连接与游标 #

手动调用close()容易遗漏，用with语句可以自动释放资源，同时确保异常时的事务处理更可靠：

merge_query = """
MERGE INTO sql_table_name AS Target
USING (
    VALUES {}
) AS Source (transaction_year, month_num, month_name, price_nt)
ON Target.transaction_year = Source.transaction_year 
AND Target.month_num = Source.month_num
WHEN MATCHED AND (Target.month_name != Source.month_name OR Target.price_nt != Source.price_nt) THEN
    UPDATE SET Target.month_name = Source.month_name, Target.price_nt = Source.price_nt
WHEN NOT MATCHED THEN
    INSERT (transaction_year, month_num, month_name, price_nt) VALUES (Source.transaction_year, Source.month_num, Source.month_name, Source.price_nt);
""".format(','.join(['(?,?,?,?)' for _ in range(len(data))]))

params = [item for sublist in data for item in sublist]

try:
    with obj_cnxn:
        with obj_cnxn.cursor() as obj_crsr:
            obj_crsr.execute(merge_query, params)
except Exception as e:
    print(f"执行失败: {e}")
    print("事务已回滚")

注意：with obj_cnxn会自动处理提交/回滚——如果代码块正常执行则提交，发生异常则回滚，无需手动调用commit()或rollback()。

2. 对动态表名/字段名做白名单校验 #

如果你的表名或字段名是动态生成的（而非硬编码），绝对不能直接拼接字符串，必须通过白名单校验确保只有合法的名称被使用：

# 示例：表名白名单
ALLOWED_TABLES = {"sql_table_name", "another_valid_table"}
target_table = "sql_table_name"  # 假设这是动态传入的值

if target_table not in ALLOWED_TABLES:
    raise ValueError(f"非法表名: {target_table}")

merge_query = f"""
MERGE INTO {target_table} AS Target
...  # 其余SQL逻辑不变
"""

3. 提前校验数据类型与格式 #

在传入数据库前，对数据进行校验，确保字段类型匹配（比如transaction_year是整数，price_nt是浮点数），既避免数据库报错，也能过滤恶意构造的数据：

import re

def validate_data(row):
    year, month_num, month_name, price = row
    # 校验年份是整数
    if not isinstance(year, int):
        raise ValueError(f"无效年份: {year}")
    # 校验月份格式（比如M开头+两位数字）
    if not re.match(r'^M\d{2}$', month_num):
        raise ValueError(f"无效月份格式: {month_num}")
    # 校验价格是数字类型
    if not isinstance(price, (int, float)):
        raise ValueError(f"无效价格: {price}")
    return row

# 对所有数据进行校验
validated_data = [validate_data(row) for row in data]
params = [item for sublist in validated_data for item in sublist]

4. 关于命名参数的适配 #

pyodbc支持命名参数（如:transaction_year），但对于批量MERGE场景，位置参数?更简洁高效。如果要使用命名参数，单条数据的写法示例如下：

# 单条数据的命名参数写法
single_row = (2023, "M12", "December", 541.44)
merge_query = """
MERGE INTO sql_table_name AS Target
USING (
    VALUES (:year, :month_num, :month_name, :price)
) AS Source (transaction_year, month_num, month_name, price_nt)
...  # 其余逻辑不变
"""
obj_crsr.execute(merge_query, year=single_row[0], month_num=single_row[1], month_name=single_row[2], price=single_row[3])

但批量场景下，位置参数的写法依然是最优选择。

核心安全原则总结 #

• 永远使用参数化查询：绝对不要将用户可控数据直接拼接进SQL字符串
• 避免动态拼接表名/字段名：如果必须动态生成，一定要用白名单校验
• 严格校验输入数据：确保数据类型、格式符合业务规则
• 规范资源管理：用上下文管理器处理数据库连接与游标，避免资源泄漏

内容的提问来源于stack exchange，提问作者LT_AKR