Python操作MySQL怎么防止SQL注入：参数化查询最佳实践

在 Python 操作 MySQL 的开发过程中，彻底杜绝 SQL 注入的核心原则非常简单且明确：严格将 SQL 代码逻辑与用户输入数据分离。无论使用 PyMySQL 还是官方驱动 mysql-connector-python，最有效、最标准的方式就是始终坚持使用数据库驱动提供的参数化查询（Parameterized Query），通过统一的 %s 占位符传递参数，绝不能直接使用字符串拼接或格式化操作来组装 SQL 语句。

一、SQL 注入的根源：代码与数据的混淆

很多开发者在初学数据库操作时，习惯使用 Python 的字符串格式化工具（例如 f-string、str.format() 或 + 拼接）来拼装查询语句。这种写法将外部传入的未清洗字符串直接嵌入到了 SQL 文本中，使得外部输入获得了改变 SQL 语法树结构的可能。

# 错误写法：通过 f-string 拼接 SQL 语句
username = request.form.get("username")  # 用户输入: ' OR '1'='1
password = request.form.get("password")  # 用户输入: ' OR '1'='1

sql = f"SELECT id, username FROM users WHERE username = '{username}' AND password = '{password}'"
cursor.execute(sql)

参数化查询之所以能够从根源上防御此类攻击，是因为底层通信机制中将 SQL 语法骨架与字面量数据作为独立部分处理。MySQL 驱动要么使用服务端的预编译协议（Prepared Statement），要么在客户端对参数进行精准严格的类型转义并作为纯值传递，使得任何恶意符号都仅被当作字面量文本，绝不参与语法分析。

二、参数化查询的标准操作步骤

在实际编码中，实现参数化查询并不需要复杂的语法，只需要规范遵循驱动的执行规范。下面以常见的 PyMySQL 为例，完整演示标准四步法：

import pymysql

# 步骤 1：安全建立数据库连接与游标
connection = pymysql.connect(
    host="127.0.0.1",
    user="app_user",
    password="your_strong_password",
    database="production_db",
    charset="utf8mb4",
    cursorclass=pymysql.cursors.DictCursor
)

try:
    with connection.cursor() as cursor:
        # 步骤 2：编写包含 %s 占位符的 SQL 模板（注意：%s 两侧绝不要加单引号）
        sql_template = "SELECT id, username, email FROM users WHERE username = %s AND status = %s"
        
        # 步骤 3：准备参数元组（或列表），与占位符数量严格一一对应
        input_username = "admin_user"
        input_status = 1
        query_params = (input_username, input_status)
        
        # 步骤 4：将 SQL 和参数作为独立实参传给 cursor.execute()
        cursor.execute(sql_template, query_params)
        
        # 获取查询结果
        result = cursor.fetchall()
        print("查询结果（示例）：", result)
finally:
    connection.close()

三、常见误区与深水区应对方案

虽然参数化查询的概念容易理解，但在遇到稍复杂的实际业务场景时，很多开发者容易因为不清楚驱动规范或缺乏灵活性而重新走回字符串拼接的老路。

误区 1：在占位符两侧加引号

初学者常常误以为数字用 %s，字符串就应该写成 '%s'。这是严重的错误理解。

• 在 PyMySQL 和 mysql-connector 中，占位符一律使用 %s，代表的是一个语法占位符，而不是 Python 原生的字符串替换指令。
• 如果在 SQL 中写了 WHERE username = '%s'，数据库引擎会将其解析为一个字面值就是 %s 两个字符的普通字符串常量，参数根本无法被安全替换，或者导致类型解析失败。

误区 2：误以为单个参数可以直接传变量

# 常见隐蔽报错：
user_id = 1001
cursor.execute("SELECT * FROM users WHERE id = %s", user_id)  # 错误！TypeError

cursor.execute(sql, args) 的第二个参数必须是一个序列（元组、列表）或字典。当只有一个参数时，如果传入字符串或整数，驱动内部尝试遍历时会报错。正确写法必须加逗号表示单元素元组：(user_id,) 或者使用列表 [user_id]。

误区 3：IN 查询多项动态列表

在实际业务中，我们经常需要处理类似 WHERE id IN (...) 的查询，列表项由前端动态传入。由于 MySQL 驱动的 %s 占位符只代表一个标量值，不能直接传入一个整个列表，很多开发者在这里慌乱之下又改用拼接。

安全解法是：动态生成对应数量的占位符字符串，参数列表解包传入：

# 安全处理动态 IN 列表
user_ids = [101, 102, 103, 108]

if not user_ids:
    results = []
else:
    # 动态构造对应数量的 %s，结果如: %s, %s, %s, %s
    placeholders = ", ".join(["%s"] * len(user_ids))
    sql = f"SELECT id, username FROM users WHERE id IN ({placeholders})"
    
    # 仍将参数列表通过 execute 第二个参数安全传递
    cursor.execute(sql, user_ids)
    results = cursor.fetchall()

误区 4：表名、列名、排序方向动态指定

参数化查询仅适用于传递数据值（字面量），不能用于 SQL 结构成分，例如表名（TABLE）、列名（COLUMN）或排序方向（ASC/DESC）。如果在表名处写 SELECT * FROM %s，执行时必定报错。

当业务确实需要由用户指定表名或排序字段时，必须采用严格白名单校验（Whitelist Validation）：

# 动态排序列的安全白名单设计
ALLOWED_SORT_FIELDS = {
    "created_at": "created_at",
    "score": "score",
    "name": "username"
}
ALLOWED_ORDERS = {"asc": "ASC", "desc": "DESC"}

# 接收用户输入
user_sort = request.args.get("sort", "created_at")
user_order = request.args.get("order", "desc").lower()

# 严格白名单校验：只允许已知枚举值通过，杜绝任意字符
if user_sort not in ALLOWED_SORT_FIELDS or user_order not in ALLOWED_ORDERS:
    raise ValueError("非法排序参数")

safe_column = ALLOWED_SORT_FIELDS[user_sort]
safe_direction = ALLOWED_ORDERS[user_order]

# 白名单内字段安全嵌入结构，过滤条件仍然使用参数化
sql = f"SELECT id, username FROM users WHERE status = %s ORDER BY {safe_column} {safe_direction}"
cursor.execute(sql, (1,))

四、批量写入与事务安全最佳实践

除了单条记录的查询，在批量数据导入或新增场景下，千万不要使用循环加拼接的方式构造庞大的 INSERT INTO users VALUES (...), (...) 语句。循环拼接不仅代码脆弱，还容易触发缓冲区溢出或注入风险。

推荐的最佳实践是使用游标提供的 cursor.executemany(sql, param_list) 方法，并结合事务安全控制：

# 批量安全写入最佳实践
insert_sql = "INSERT INTO users (username, email, role, status) VALUES (%s, %s, %s, %s)"

# 构造数据列表，每个元素都是与占位符对齐的元组
records_to_insert = [
    ("alice_w", "alice@example.com", "editor", 1),
    ("bob_k", "bob@example.com", "viewer", 1),
    ("charlie_z", "charlie@example.com", "admin", 0)
]

try:
    with connection.cursor() as cursor:
        # 一次性执行批量插入，驱动会自动批量优化且严格保证数据安全
        rows_affected = cursor.executemany(insert_sql, records_to_insert)
        print("影响行数（示例）：", rows_affected)
    
    # 事务提交
    connection.commit()
except Exception as e:
    # 异常时坚决回滚
    connection.rollback()
    print("写入异常，事务已回滚:", e)

使用 executemany 时，驱动内部会复用模板并依次安全转义，不仅杜绝了 SQL 注入，还能在底层进行性能批量优化。

五、防注入核对清单与工程防御体系

防范 SQL 注入是一项团队工程实践，除了在个人编写代码时严格遵守参数化规则外，还需要在项目规范与审查流程中设立防御网：

坚持在日常开发中将这些参数化实践落实到每一行 SQL 调用中，配合代码审查与最小权限原则，就能在 Python 开发中从根本上构建起稳固的防注入屏障。