Python操作数据库怎么防止SQL注入？参数化查询最佳实践

何川

发布时间：2026-11-11 | 3人浏览过

原创

详细讲解Python操作数据库时防止SQL注入的根本方法与最佳实践，深入剖析参数化查询在sqlite3、PyMySQL与psycopg2中的具体用法、边缘场景处理及工程规范。

Python 数据库安全指南

在 Python 中防止 SQL 注入的唯一正解是使用参数化查询（Parameterized Queries）。无论使用内置的 sqlite3 还是第三方的 PyMySQL、psycopg2，都必须由数据库驱动程序负责参数的绑定与转义，严禁用 Python 原生的字符串拼接（如 f-string、% 格式化或 + 号）去组装 SQL 语句。本文将从 SQL 注入的触发机理展开，通过真实代码示例拆解参数化查询的标准写法、常见驱动的语法差异，以及动态表名、IN 子句等多值场景下的安全处理方案。

Python操作数据库怎么防止SQL注入？参数化查询最佳实践 — 文章核心结构图：静态 SQL 模版与外部输入参数在驱动层实施物理分离

一、SQL 注入的根本成因与典型危险写法

SQL 注入并非因为外部输入包含了“奇怪的特殊符号”，而是因为程序将不可信的用户数据直接拼合进了 SQL 语句文本中，导致数据库词法分析器（Lexer）将用户提供的数据识别为了可执行的指令关键字。

在关系型数据库处理查询时，首先需要解析 SQL 语句并构建抽象语法树（AST）。如果采用字符串拼接，外部数据在解析前就已经与原始命令混杂在一起，攻击者只需巧妙构造闭合符号（如单引号 '、注释符 -- 或逻辑词 OR），就能彻底改变原始查询的语法树拓扑结构。

常见的致命错误写法（反面教材）

在日常代码审查中，以下三种拼接方式最为常见，均属于极度危险的操作：

# 错误 1：使用 Python 3.6+ 的 f-string 字符串插值
cursor.execute(f"SELECT * FROM users WHERE username = '{username}' AND password = '{password}'")

# 错误 2：使用传统的 % 格式化运算符
cursor.execute("SELECT * FROM users WHERE username = '%s'" % (username,))

# 错误 3：使用 str.format() 方法或加号拼接
cursor.execute("SELECT * FROM users WHERE username = '{}'".format(username))
cursor.execute("SELECT * FROM users WHERE username = '" + username + "'")

当攻击者在登录界面的用户名输入框中提交 admin' OR '1'='1 时，上述代码拼出的最终 SQL 会变成：

SELECT * FROM users WHERE username = 'admin' OR '1'='1' AND password = '...'

因为 OR '1'='1' 永远为真，并且由于操作符优先级关系，验证逻辑被彻底绕过，攻击者无需知道密码即可登录首个管理员账户。

二、参数化查询的工作机理：数据与指令的物理分离

参数化查询（在底层常对应预编译语句 Prepared Statements）的本质是“两阶段处理”：

编译阶段（固定语法树）：应用程序先将带有占位符的纯静态 SQL 发送给数据库（或驱动）。数据库对其进行语法分析、词法分析和优化，生成不可变的高效执行计划。此时，语法树的结构已经被焊死，哪里是表名、哪里是条件判断早已确定。
绑定阶段（填充字面量）：外部变量作为独立的参数包单独传输给数据库。数据库引擎只把这些参数当作纯粹的值（Literal Value）填入对应的数据槽位，绝不重新解析其中的任何语法结构。

在参数化查询模式下，即便用户输入的内容包含 ' OR '1'='1，数据库也只会去查找数据库字段 username 刚好等于字符串 "admin' OR '1'='1" 的用户，这串输入绝无可能变成控制流程的布尔运算符。

三、主流 Python 数据库驱动的实战规范

Python 社区制定了统一的数据库接口规范 Python DB-API 2.0 (PEP 249)。虽然绝大多数主流驱动都遵循该规范，但不同驱动在占位符风格（paramstyle）上存在细微差别。如果占位符写错，程序要么抛出语法异常，要么退化为无效查询。

数据库类型	驱动库名	规范属性 (paramstyle)	占位符语法	传递参数类型
SQLite	`sqlite3` (标准库)	`qmark` / `named`	`?` 或 `:name`	元组 (tuple) 或字典 (dict)
MySQL / MariaDB	`PyMySQL` / `mysql-connector`	`format` / `pyformat`	`%s`	元组 (tuple) 或列表 (list)
PostgreSQL	`psycopg2` / `psycopg`	`pyformat`	`%s` 或 `%(key)s`	元组 (tuple) 或字典 (dict)

1. SQLite (sqlite3) 标准写法

Python 内置的 sqlite3 模块使用问号 ? 作为位置占位符，或者使用冒号前缀 :name 作为具名占位符：

import sqlite3

def query_user_sqlite(user_email: str):
    with sqlite3.connect("app.db") as conn:
        cursor = conn.cursor()
        
        # 正确做法：使用 ? 作为占位符，参数通过第二个入参元组传入
        sql = "SELECT id, username, email FROM users WHERE email = ? AND is_active = 1"
        cursor.execute(sql, (user_email,))  # 注意：单元素元组末尾必须带逗号
        
        user = cursor.fetchone()
        return user

def query_user_by_dict_sqlite(user_id: int):
    with sqlite3.connect("app.db") as conn:
        cursor = conn.cursor()
        
        # 亦可使用具名占位符，第二参数传入字典，语义更直观
        sql = "SELECT id, username FROM users WHERE id = :uid"
        cursor.execute(sql, {"uid": user_id})
        return cursor.fetchone()

避坑提醒：单元素元组必须带有逗号

在 Python 中，括号包裹单个对象 (user_email) 的类型仍是 str 而非 tuple。若写成 cursor.execute(sql, (user_email))，驱动会把字符串迭代展开成一个个单字符，引发参数数量不匹配错误（Incorrect number of bindings supplied）。必须写作 (user_email,)。

2. MySQL (PyMySQL) 标准写法

很多新手在 PyMySQL 中看到 %s，误以为这是 Python 的字符串格式化，从而写出 sql % (param)。请务必区分：这里的 %s 是传递给数据库驱动的协议占位符，必须作为 execute() 的独立参数传递！

import pymysql

def update_user_status(user_id: int, status: str):
    connection = pymysql.connect(
        host="127.0.0.1", user="root", password="pwd", database="prod_db"
    )
    try:
        with connection.cursor() as cursor:
            # 正确做法：SQL 中仅写纯 %s，千万不要在 %s 外层包裹单引号！
            sql = "UPDATE users SET status = %s, updated_at = NOW() WHERE id = %s"
            cursor.execute(sql, (status, user_id))
            
        connection.commit()
    finally:
        connection.close()

避坑提醒：不要给 %s 占位符加引号

如果写成 WHERE status = '%s'，数据库驱动会将 '%s' 视作一个字面值字符串常量 "%s"，不再将其作为占位变量解析，进而导致参数注入失败甚至语法错误。无论字段是字符串、整数还是时间类型，占位符一律直接裸写 %s 或 ?。

3. PostgreSQL (psycopg2) 标准写法

psycopg2 支持位置占位符 %s，也支持字典映射占位符 %(name)s：

import psycopg2

def insert_new_log(event_type: str, message: str, operator_id: int):
    with psycopg2.connect("dbname=app user=postgres") as conn:
        with conn.cursor() as cur:
            sql = """
                INSERT INTO audit_logs (event_type, message, operator_id)
                VALUES (%(evt)s, %(msg)s, %(op_id)s)
            """
            params = {
                "evt": event_type,
                "msg": message,
                "op_id": operator_id
            }
            cur.execute(sql, params)

四、高频陷阱与边缘场景的正确处理

很多开发者知道基本查询要用参数化，但在遇到“动态表名”、“排序字段（ORDER BY）”或“IN 范围查询”时，由于不知道参数化如何生效，便重新退回到了危险的字符串拼接。针对这些边缘场景，必须采取分类防御策略。

1. 陷阱：动态表名、列名与排序字段不能使用参数化

必须明确一个根本原则：数据库的参数化占位符只能代替“数据值（Data Value）”，绝不能代替 SQL 的“元数据/标识符（Identifier）”。表名、列名、操作符、ASC / DESC 等属于 SQL 语法结构的一部分，无法被预编译为参数槽位。

# 严重语法错误：驱动会报错，因为参数化无法替换表名与列名
cursor.execute("SELECT * FROM ? WHERE ? = ?", ("users", "status", "active"))

安全解决之道：严格白名单映射（Whitelist Mapping）

面对用户传入的排序列或查询表名，决不允许直接拼入 SQL。正确的做法是在代码中维护一个只读的安全字典或集合，将用户输入转换为受控的固定标识符：

def get_sorted_users(cursor, sort_by: str, order: str = "asc"):
    # 1. 建立允许排序的字段白名单映射
    ALLOWED_SORT_COLUMNS = {
        "name": "username",
        "created": "created_at",
        "score": "user_score"
    }
    
    # 2. 校验并获取安全列名（找不到则回退到安全默认值）
    column = ALLOWED_SORT_COLUMNS.get(sort_by)
    if not column:
        raise ValueError("不合法的排序字段")
        
    # 3. 严格限定排序方向
    direction = "DESC" if order.lower() == "desc" else "ASC"
    
    # 4. 经过白名单强校验后的受控拼接，配合参数化传递过滤条件
    safe_sql = f"SELECT id, username, created_at FROM users WHERE is_active = ? ORDER BY {column} {direction}"
    cursor.execute(safe_sql, (1,))
    return cursor.fetchall()

在 PostgreSQL 中，还可以使用 psycopg2.sql 模块提供的 Identifier 工具自动转义标识符：

from psycopg2 import sql

query = sql.SQL("SELECT * FROM {table} WHERE id = %s").format(
    table=sql.Identifier(user_input_table)
)
cursor.execute(query, (target_id,))

2. 陷阱：IN 子句的多值动态查询

在处理 WHERE id IN (1, 2, 3) 需求时，初学者常常直接把列表或元组传入占位符：

# 错误写法：占位符不会自动将元组展开为逗号分隔序列
ids = [101, 102, 103]
cursor.execute("SELECT * FROM users WHERE id IN (?)", (ids,))  # 报错！

安全解决之道：根据参数长度动态生成安全占位符

生成若干个 ? 只是动态拼接了“无害的占位符号”，实际的数据列表依然作为独立参数绑定，因此完全安全：

def query_users_by_id_list(cursor, ids: list[int]):
    if not ids:
        return []
    
    # 动态生成与列表长度匹配的问号占位符：?, ?, ?
    placeholders = ", ".join(["?"] * len(ids))
    sql = f"SELECT id, username FROM users WHERE id IN ({placeholders})"
    
    # 真实数据依然通过第二个参数传递
    cursor.execute(sql, tuple(ids))
    return cursor.fetchall()

3. 批量写入的最佳实践：executemany

需要插入或更新大量记录时，切勿通过循环拼装大 SQL 字符串，而应使用驱动原生的 executemany()。驱动底层会高效复用预编译模板：

user_records = [
    ("alice", "alice@example.com", 28),
    ("bob", "bob@example.com", 32),
    ("charlie", "charlie@example.com", 24)
]

sql = "INSERT INTO users (username, email, age) VALUES (?, ?, ?)"
cursor.executemany(sql, user_records)
connection.commit()

五、ORM 框架下的防注入注意事项

现代 Python 常用 ORM（如 SQLAlchemy、Django ORM）默认采用参数化查询，绝大部分通过模型 API 调用的查询均天然免疫 SQL 注入：

# SQLAlchemy ORM 表达式：框架底层自动生成参数化绑定，安全
session.query(User).filter(User.username == user_input).first()

# Django ORM 查询：框架底层使用参数化执行，安全
User.objects.filter(username=user_input)

但必须高度警惕：ORM 并不等于免死金牌！当在 ORM 中编写原生 SQL 语句（Raw SQL）时，如果开发者疏忽大意再次使用字符串拼接，依然会导致 SQL 注入。

在 SQLAlchemy 中编写原生 SQL 时，必须使用 text() 并显式绑定具名参数：

from sqlalchemy import text

# 致命危险：在 text() 内部使用 f-string 拼接原生 SQL
session.execute(text(f"SELECT * FROM users WHERE email = '{user_email}'"))

# 安全推荐：使用冒号命名参数，在 params 字典中绑定数据
safe_stmt = text("SELECT * FROM users WHERE email = :email AND status = :status")
result = session.execute(safe_stmt, {"email": user_email, "status": "active"})

六、总结与团队工程落地建议

保障 Python 数据库访问安全的本质在于建立可靠的开发与审查基线：

绝对禁用字符串拼接：禁止任何形式的 f-string、%、format()、+ 用于构造 SQL 语句中的数据值。
参数与逻辑严格分离：牢记 DB-API 规范，所有外部传入变量必须通过 cursor.execute(sql, params) 作为第二个参数传入。
元数据实施白名单：表名、字段名、排序关键字等不能参数化的标识符，必须通过显式白名单校验，杜绝任何外部输入直达 SQL。
引入自动化静态扫描：在 CI/CD 流程中集成针对 Python 安全漏洞的静态检查工具（例如 Bandit）。Bandit 的 B608 规则能够在代码提交前自动识别硬编码或格式化拼接的 SQL 语句，并在流水线阶段直接阻断高危代码合入主干。