Python sqlite3参数化查询示例

Python sqlite3参数化查询的核心就是使用占位符（? 或命名占位符）来构建SQL语句，然后将数据作为参数单独传递给执行方法。直接拼接字符串（如 f"SELECT * FROM users WHERE id = {user_id}"）是极其危险的做法，会引发SQL注入攻击。正确做法是使用cursor.execute("SELECT * FROM users WHERE id = ?", (user_id,))，让sqlite3驱动负责安全地处理和转义参数。

为什么必须使用参数化查询？杜绝SQL注入

SQL注入是通过将恶意SQL代码插入到查询输入中，从而欺骗服务器执行非预期命令的攻击手段。假设你有一个登录查询，通过字符串拼接构建："SELECT * FROM users WHERE username = '" + username + "' AND password = '" + password + "'"。如果用户在用户名输入“admin'--”，查询就会变成“SELECT * FROM users WHERE username = 'admin'--' AND password = '...'”。“--”在SQL中是注释符，这意味着密码检查被完全绕过，攻击者可以直接以管理员身份登录。而参数化查询将数据与指令分离，数据库驱动会确保输入的数据永远只被当作数据处理，而不会成为可执行的SQL代码的一部分，从根本上杜绝了此类攻击。

两种占位符语法：问号占位符与命名占位符

Python的sqlite3模块主要支持两种参数化查询的语法。第一种是使用问号“？”作为占位符。这种方法按位置传递参数，参数必须是一个元组或列表。当参数只有一个时，必须写成单元素元组的形式，即（value,），后面的逗号不能省略。

import sqlite3

conn = sqlite3.connect('example.db')
cursor = conn.cursor()

# 创建表
cursor.execute('''CREATE TABLE IF NOT EXISTS employees
                  (id INTEGER PRIMARY KEY, name TEXT, department TEXT, salary REAL)''')

# 插入数据 - 使用 ? 占位符
employee_data = ('张三', '技术部', 8500.00)
cursor.execute("INSERT INTO employees (name, department, salary) VALUES (?, ?, ?)", employee_data)

# 查询数据
dept = '技术部'
cursor.execute("SELECT * FROM employees WHERE department = ?", (dept,))
rows = cursor.fetchall()
for row in rows:
    print(row)

conn.commit()
conn.close()

第二种是使用命名占位符，格式为“:name”。这种方式使用字典来传递参数，代码可读性更强，尤其适用于参数众多或需要多次重复使用的场景。

import sqlite3

conn = sqlite3.connect('example.db')
cursor = conn.cursor()

# 使用命名占位符插入数据
employee_dict = {'emp_name': '李四', 'emp_dept': '市场部', 'emp_salary': 7200.00}
cursor.execute("INSERT INTO employees (name, department, salary) VALUES (:emp_name, :emp_dept, :emp_salary)", employee_dict)

# 使用命名占位符进行复杂查询
query_params = {'min_salary': 8000.0, 'target_dept': '技术部'}
cursor.execute("""
    SELECT name, salary FROM employees
    WHERE department = :target_dept AND salary > :min_salary
    ORDER BY salary DESC
""", query_params)

for row in cursor.fetchall():
    print(f"姓名：{row[0]}, 薪资：{row[1]}")

conn.commit()
conn.close()

executemany()：高效批量操作

当你需要向数据库中插入或更新大量记录时，逐条执行execute()会非常低效。sqlite3提供了executemany()方法，它接受一个SQL语句和一个包含多组参数的序列（列表或元组），在一次调用中执行所有操作，这能显著提升性能。

import sqlite3

conn = sqlite3.connect('example.db')
cursor = conn.cursor()

# 批量插入数据
new_employees = [
    ('王五', '销售部', 6500.00),
    ('赵六', '技术部', 9000.00),
    ('钱七', '人事部', 6000.00)
]

cursor.executemany("INSERT INTO employees (name, department, salary) VALUES (?, ?, ?)", new_employees)

# 批量更新数据
salary_adjustments = [
    (7500.00, '销售部'),
    (9500.00, '技术部')  # 假设为这两个部门调薪
]
cursor.executemany("UPDATE employees SET salary = ? WHERE department = ?", salary_adjustments)

conn.commit()
print(f"受影响的行数：{cursor.rowcount}")
conn.close()

查询中的高级参数化技巧

参数化查询不仅用于WHERE子句，它可以用于SQL语句中任何需要动态值的地方。一个常见的误区是试图将表名或列名等SQL标识符参数化，这是不允许的。占位符只能代表值（字符串、数字等）。对于动态标识符，必须在Python层面进行安全的过滤和拼接。

import sqlite3

def safe_query_by_column(db_path, column_name, column_value):
    """一个安全的、允许动态列名查询的函数示例"""
    # 预先定义允许查询的列名白名单，防止任意列名注入
    allowed_columns = ['name', 'department', 'salary']
    if column_name not in allowed_columns:
        raise ValueError(f"不允许查询列：{column_name}")

    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
    # 列名通过安全的字符串拼接，值使用参数化查询
    query_sql = f"SELECT * FROM employees WHERE {column_name} = ?"
    cursor.execute(query_sql, (column_value,))
    results = cursor.fetchall()
    conn.close()
    return results

# 安全调用
print(safe_query_by_column('example.db', 'department', '技术部'))

结合上下文管理器与连接池模式

在实际项目中，为了确保数据库连接被正确关闭，推荐使用“with”上下文管理器。从Python 3.4开始，sqlite3连接对象支持上下文管理器协议，这可以自动提交事务或在发生异常时回滚。

import sqlite3
from contextlib import closing

# 使用连接和游标的上下文管理器
def add_employee_safe(emp_info):
    try:
        with sqlite3.connect('example.db', timeout=10) as conn:
            # 设置行工厂，返回字典形式的结果，更易用
            conn.row_factory = sqlite3.Row
            with closing(conn.cursor()) as cursor: # 使用closing确保游标关闭
                cursor.execute("""
                    INSERT INTO employees (name, department, salary)
                    VALUES (:name, :dept, :sal)
                """, emp_info)
            # with块结束时会自动commit，如果发生异常则自动rollback
    except sqlite3.Error as e:
        print(f"数据库错误：{e}")
        # 根据业务需求决定是否重新抛出异常

# 调用
add_employee_safe({'name': '孙八', 'dept': '财务部', 'sal': 6800.00})

# 查询并获取字典
with sqlite3.connect('example.db') as conn:
    conn.row_factory = sqlite3.Row
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM employees WHERE salary > ?", (7000,))
    for row in cursor.fetchall():
        # 可以通过列名访问
        print(dict(row))  # 输出：{'id': 1, 'name': '张三', ...}

性能考量与最佳实践总结

1. 始终使用参数化查询：这是铁律，没有任何例外。无论是来自用户输入、配置文件还是API接口的数据，都必须参数化。
2. 利用预编译语句（Prepared Statements）的优势：当你重复执行同一条SQL语句（仅参数不同）时，如在一个循环中插入数据，应该先准备语句。虽然Python sqlite3的execute()在内部有一定缓存，但在高性能场景下，显式使用cursor.execute(sql, params)并复用该SQL字符串，能让数据库驱动更好地重用查询计划。
3. 管理好事务：对于批量操作，务必显式控制事务。将多个executemany()或execute()放在一个begin commit块中，比自动提交模式快数十倍。
4. 警惕数据类型：sqlite3是动态类型，但Python类型（如None, int, float, str, bytes）到SQLite类型（NULL, INTEGER, REAL, TEXT, BLOB）的映射是清晰的。确保传递的参数类型符合预期，避免不必要的类型转换开销。
5. 关闭连接：虽然上下文管理器能自动处理，但在长生命周期应用或Web应用中，要确保每个请求结束后都及时关闭数据库连接，避免资源泄露。

掌握Python sqlite3参数化查询，不仅仅是学会一种语法，更是树立起牢固的数据库安全与编程规范意识。它保证了应用的安全性、提升了代码的健壮性，并为处理更复杂的数据操作奠定了坚实基础。

服务指南

帮助中心