Python sqlite3.Cursor.executemany 方法

上次修改时间：2025 年 4 月 15 日

本综合指南探讨了 Python 的 sqlite3.Cursor.executemany 方法，用于在 SQLite 数据库中进行高效的批量操作。我们将介绍基本用法、参数处理、性能考量和实际示例。

基本定义

executemany 方法针对序列中的所有参数序列执行参数化的 SQL 命令。它经过优化，可以通过单个方法调用插入或更新多个行。

主要特点：它接受一个 SQL 模板和一个参数的可迭代对象，为每个参数集执行语句，并且对于批量操作来说，比多次调用 execute 更有效。

基本批量插入

此示例演示了 executemany 的最简单用法，只需一次调用即可将多行插入到表中。

basic_executemany.py

import sqlite3

with sqlite3.connect('products.db') as conn:
    cursor = conn.cursor()
    cursor.execute('''CREATE TABLE IF NOT EXISTS products
                     (id INTEGER PRIMARY KEY, name TEXT, price REAL)''')
    
    products = [
        ('Laptop', 999.99),
        ('Mouse', 19.99),
        ('Keyboard', 49.99),
        ('Monitor', 199.99)
    ]
    
    cursor.executemany("INSERT INTO products (name, price) VALUES (?, ?)", products)
    conn.commit()

该示例创建一个 products 表，并通过一次 executemany 调用插入四个产品。 products 列表中的每个元组对应于插入的一行。

使用 with 确保正确的资源清理。连接在成功完成时自动提交，或在出现异常时回滚。

使用命名参数批量插入

此示例演示了如何将命名参数与 executemany 结合使用，以便在处理多个列时获得更清晰的代码。

named_params.py

import sqlite3

with sqlite3.connect('employees.db') as conn:
    cursor = conn.cursor()
    cursor.execute('''CREATE TABLE IF NOT EXISTS employees
                     (id INTEGER PRIMARY KEY, name TEXT, 
                      department TEXT, salary REAL)''')
    
    employees = [
        {'name': 'Alice', 'department': 'HR', 'salary': 55000},
        {'name': 'Bob', 'department': 'IT', 'salary': 65000},
        {'name': 'Charlie', 'department': 'Finance', 'salary': 70000}
    ]
    
    cursor.executemany(
        "INSERT INTO employees (name, department, salary) VALUES (:name, :department, :salary)",
        employees
    )
    conn.commit()

这里我们使用带有命名占位符 (:name) 的字典参数。这使得代码更具可读性和可维护性，尤其是在有很多列的情况下。

字典键必须与 SQL 语句中的命名占位符匹配。使用命名参数时，顺序无关紧要。

批量更新

executemany 不仅适用于插入 - 它同样适用于批量更新。此示例一次操作即可更新多个记录。

batch_update.py

import sqlite3

with sqlite3.connect('inventory.db') as conn:
    cursor = conn.cursor()
    cursor.execute('''CREATE TABLE IF NOT EXISTS inventory
                     (id INTEGER PRIMARY KEY, item TEXT, quantity INTEGER)''')
    
    # Initial data
    cursor.executemany(
        "INSERT INTO inventory (item, quantity) VALUES (?, ?)",
        [('Apples', 50), ('Oranges', 30), ('Bananas', 40)]
    )
    
    # Update quantities
    updates = [
        (35, 'Apples'),
        (25, 'Oranges'),
        (45, 'Bananas')
    ]
    
    cursor.executemany(
        "UPDATE inventory SET quantity = ? WHERE item = ?",
        updates
    )
    conn.commit()

此示例首先插入一些库存项目，然后更新它们的数量。更新参数以元组列表的形式提供，与 SQL 模板匹配。

请注意，更新元组中的参数顺序必须与 SQL 语句中的占位符匹配（先是 quantity，然后是 item）。

使用生成器的大型批量插入

对于非常大的数据集，您可以将生成器与 executemany 结合使用，以避免一次将所有数据加载到内存中。

generator_executemany.py

import sqlite3

def generate_data(num_records):
    for i in range(1, num_records + 1):
        yield (f'Product_{i}', i * 10.0)

with sqlite3.connect('large_data.db') as conn:
    cursor = conn.cursor()
    cursor.execute('''CREATE TABLE IF NOT EXISTS large_products
                     (id INTEGER PRIMARY KEY, name TEXT, price REAL)''')
    
    # Insert 10,000 records without loading all into memory
    cursor.executemany(
        "INSERT INTO large_products (name, price) VALUES (?, ?)",
        generate_data(10000)
    )
    conn.commit()

generate_data 函数一次生成一条记录，允许 executemany 处理它们，而无需将所有 10,000 条记录存储在内存中。

对于大型数据集，这种方法在内存使用方面效率很高。生成器按需创建数据，因为 executemany 会处理每条记录。

使用 executemany 进行事务处理

此示例演示了 executemany 的事务行为，显示了当批处理中的一部分失败时如何处理错误。

transaction_handling.py

import sqlite3

with sqlite3.connect('transactions.db') as conn:
    cursor = conn.cursor()
    cursor.execute('''CREATE TABLE IF NOT EXISTS accounts
                     (id INTEGER PRIMARY KEY, name TEXT, balance REAL)''')
    
    # Initial data
    cursor.execute("INSERT INTO accounts (name, balance) VALUES ('Alice', 1000)")
    cursor.execute("INSERT INTO accounts (name, balance) VALUES ('Bob', 500)")
    conn.commit()
    
    # Batch transfer that will fail
    transfers = [
        (100, 'Alice'),  # Valid
        (600, 'Bob'),     # Invalid (Bob only has 500)
        (200, 'Alice')    # Would be valid but won't execute
    ]
    
    try:
        cursor.executemany(
            "UPDATE accounts SET balance = balance - ? WHERE name = ?",
            transfers
        )
        conn.commit()
    except sqlite3.Error as e:
        print("Error occurred:", e)
        conn.rollback()

该示例尝试在一次 executemany 调用中进行三项帐户更新。第二次更新失败（资金不足），导致整个批处理失败。

事务已回滚，使数据库保持其原始状态。这表明 executemany 是原子性的 - 要么所有操作都成功，要么都不成功。

性能比较

此示例将 executemany 的性能与单个 execute 调用进行比较，以演示效率差异。

performance_comparison.py

import sqlite3
import time

def time_executemany():
    with sqlite3.connect(':memory:') as conn:
        cursor = conn.cursor()
        cursor.execute('''CREATE TABLE test (id INTEGER, data TEXT)''')
        
        data = [(i, f'Data_{i}') for i in range(10000)]
        
        start = time.time()
        cursor.executemany("INSERT INTO test VALUES (?, ?)", data)
        conn.commit()
        return time.time() - start

def time_individual():
    with sqlite3.connect(':memory:') as conn:
        cursor = conn.cursor()
        cursor.execute('''CREATE TABLE test (id INTEGER, data TEXT)''')
        
        data = [(i, f'Data_{i}') for i in range(10000)]
        
        start = time.time()
        for item in data:
            cursor.execute("INSERT INTO test VALUES (?, ?)", item)
        conn.commit()
        return time.time() - start

print(f"executemany time: {time_executemany():.4f} seconds")
print(f"Individual execute time: {time_individual():.4f} seconds")

该示例创建一个内存数据库，并使用这两种方法计时插入 10,000 条记录。 executemany 通常明显更快。

性能差异来自减少的 Python-SQLite 往返行程以及 SQLite 中针对批量操作的内部优化。

大型操作的自定义批处理大小

对于非常大的操作，您可能需要对数据进行分块，以避免内存问题，同时仍然受益于批处理性能。

chunked_executemany.py

import sqlite3

def generate_large_data(num_records):
    for i in range(1, num_records + 1):
        yield (f'Item_{i}', i % 100, i * 1.5)

BATCH_SIZE = 1000

with sqlite3.connect('huge_data.db') as conn:
    cursor = conn.cursor()
    cursor.execute('''CREATE TABLE IF NOT EXISTS huge_table
                     (id INTEGER PRIMARY KEY, name TEXT, 
                      category INTEGER, value REAL)''')
    
    batch = []
    for record in generate_large_data(100000):
        batch.append(record)
        if len(batch) >= BATCH_SIZE:
            cursor.executemany(
                "INSERT INTO huge_table (name, category, value) VALUES (?, ?, ?)",
                batch
            )
            batch = []
    
    # Insert any remaining records
    if batch:
        cursor.executemany(
            "INSERT INTO huge_table (name, category, value) VALUES (?, ?, ?)",
            batch
        )
    
    conn.commit()

此示例以 1,000 个批处理处理 100,000 条记录。它平衡了内存使用和批处理性能优势。

当处理的数据集太大而无法在一次 executemany 调用中处理，但单个插入太慢时，此方法非常有用。

最佳实践

用于批量操作： 使用大型数据集可显著提高性能
注意事务大小： 非常大的批处理可能需要分块
首选命名参数： 在有很多列的情况下更具可读性
正确处理错误： 请记住该操作是原子性的
考虑内存使用情况： 对非常大的数据集使用生成器

资料来源

作者

我叫 Jan Bodnar，是一名充满热情的程序员，拥有丰富的编程经验。我自 2007 年以来一直在撰写编程文章。迄今为止，我撰写了 1,400 多篇文章和 8 本电子书。我拥有超过十年的编程教学经验。

列出所有 Python 教程。