Python sqlite3.Connection.executemany 方法

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

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

基本定义

executemany 方法针对序列中的所有参数序列执行参数化的 SQL 命令。它针对使用单个调用插入或更新多行进行了优化。

主要特点：它比多次 execute 调用更快，减少了 Python 到 SQLite 的往返次数，并自动维护事务完整性。该方法在 Connection 和 Cursor 对象上都可用。

基本批量插入

此示例演示了 executemany 的最简单用法，可使用单个调用将多行插入到表中。

import sqlite3

# Sample data to insert
users = [
    ('Alice', 30),
    ('Bob', 25),
    ('Charlie', 35),
    ('Diana', 28)
]

with sqlite3.connect('users.db') as conn:
    conn.execute('''CREATE TABLE IF NOT EXISTS users
                    (id INTEGER PRIMARY KEY, name TEXT, age INTEGER)''')
    
    # Insert all users with executemany
    conn.executemany("INSERT INTO users (name, age) VALUES (?, ?)", users)

executemany 方法接受带有占位符的 SQL 语句和参数序列的序列。每个内部序列都为一行插入提供值。

这种方法比单个 execute 调用更有效，因为它使用单个事务并减少了开销。连接由上下文管理器自动关闭。

使用命名参数进行批量更新

此示例演示了如何将 executemany 与命名参数一起使用以更新多行。

import sqlite3

# Sample update data
updates = [
    {'new_name': 'Alice Smith', 'user_id': 1},
    {'new_name': 'Robert Brown', 'user_id': 2},
    {'new_name': 'Charles Green', 'user_id': 3}
]

with sqlite3.connect('users.db') as conn:
    # Use named placeholders with :prefix
    conn.executemany(
        "UPDATE users SET name = :new_name WHERE id = :user_id",
        updates
    )

命名参数使 SQL 更具可读性和可维护性，尤其是在有许多列的情况下。参数序列可以是字典，将名称映射到值。

当使用不同的值更新多行，同时保持所有行在批处理中使用相同的更新逻辑时，此模式非常有用。

大型数据集插入

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

import sqlite3
import random
import string

def generate_users(count):
    """Generate random user data"""
    for i in range(count):
        name = ''.join(random.choices(string.ascii_letters, k=10))
        age = random.randint(18, 99)
        yield (name, age)

with sqlite3.connect('large.db') as conn:
    conn.execute('''CREATE TABLE IF NOT EXISTS users
                    (id INTEGER PRIMARY KEY, name TEXT, age INTEGER)''')
    
    # Insert 10,000 users without loading all into memory
    conn.executemany(
        "INSERT INTO users (name, age) VALUES (?, ?)",
        generate_users(10000)
    )

生成器 generate_users 按需生成数据，而不是在内存中创建一个大列表。 这对于巨大的数据集来说是内存高效的。

SQLite 在单个事务中有效地处理批量插入，即使存在生成器开销，这也比单个插入快得多。

批量删除操作

executemany 还可以用于批量删除操作，根据不同的条件删除多行。

import sqlite3

# IDs to delete
ids_to_delete = [(3,), (7,), (11,), (23,)]

with sqlite3.connect('users.db') as conn:
    # Create a temporary backup table
    conn.execute("CREATE TABLE IF NOT EXISTS users_backup AS SELECT * FROM users")
    
    # Delete multiple users by ID
    conn.executemany("DELETE FROM users WHERE id = ?", ids_to_delete)
    
    # Verify deletions
    remaining = conn.execute("SELECT COUNT(*) FROM users").fetchone()[0]
    print(f"Users remaining: {remaining}")

请注意单个参数的元组格式 (3,) - 每个参数序列必须是一个序列（元组、列表等），即使每行只有一个值。

此方法确保所有删除操作都以原子方式在单个事务中发生，即使某些 ID 不存在，也可以保持数据库一致性。

批量操作中的错误处理

此示例演示了使用 executemany 时的正确错误处理，以确保发生错误时的数据完整性。

import sqlite3

products = [
    (101, 'Laptop', 999.99),
    (102, 'Phone', 699.99),
    (103, 'Tablet', 399.99),
    (104, None, 199.99),  # Invalid - name cannot be NULL
    (105, 'Monitor', 249.99)
]

try:
    with sqlite3.connect('products.db') as conn:
        conn.execute('''CREATE TABLE products
                        (id INTEGER PRIMARY KEY,
                         name TEXT NOT NULL,
                         price REAL)''')
        
        conn.executemany(
            "INSERT INTO products VALUES (?, ?, ?)",
            products
        )
except sqlite3.IntegrityError as e:
    print(f"Batch insert failed: {e}")
    # Check which rows were inserted
    with sqlite3.connect('products.db') as conn:
        count = conn.execute("SELECT COUNT(*) FROM products").fetchone()[0]
        print(f"Successfully inserted {count} rows")

当 executemany 期间发生错误时，整个操作将被回滚。该示例展示了如何处理此类情况并调查部分结果。

为了更精细的错误处理，你可能需要拆分大型批次或在插入之前验证数据。

将 executemany 与自定义类型一起使用

此示例演示了如何通过注册适配器将 executemany 与自定义 Python 类型一起使用。

import sqlite3
from datetime import date

# Sample data with dates
events = [
    ('Conference', date(2023, 6, 15), date(2023, 6, 17)),
    ('Workshop', date(2023, 7, 1), date(2023, 7, 1)),
    ('Seminar', date(2023, 8, 12), date(2023, 8, 13))
]

# Register date adapter
def adapt_date(d):
    return d.isoformat()

sqlite3.register_adapter(date, adapt_date)

with sqlite3.connect('events.db') as conn:
    conn.execute('''CREATE TABLE events
                    (name TEXT, start_date TEXT, end_date TEXT)''')
    
    conn.executemany(
        "INSERT INTO events VALUES (?, ?, ?)",
        events
    )
    
    # Verify insertion
    for row in conn.execute("SELECT * FROM events"):
        print(row)

该示例为 Python 的 date 对象注册了一个自定义适配器，以便将其正确存储在 SQLite 中。 相同的适配器适用于 executemany 调用中的所有参数。

当使用非原生 SQLite 类型时，此模式非常有用，同时仍然可以从批量操作中受益。

性能比较

此示例比较了 executemany 与单个 execute 调用的性能。

import sqlite3
import time

def time_inserts(conn, data, use_executemany):
    conn.execute("CREATE TABLE temp_test (id INTEGER, value TEXT)")
    
    start = time.time()
    
    if use_executemany:
        conn.executemany("INSERT INTO temp_test VALUES (?, ?)", data)
    else:
        for row in data:
            conn.execute("INSERT INTO temp_test VALUES (?, ?)", row)
    
    conn.commit()
    elapsed = time.time() - start
    conn.execute("DROP TABLE temp_test")
    return elapsed

# Generate test data (1000 rows)
test_data = [(i, f"Value {i}") for i in range(1000)]

with sqlite3.connect(':memory:') as conn:
    # Time executemany
    tm_many = time_inserts(conn, test_data, True)
    
    # Time individual executes
    tm_single = time_inserts(conn, test_data, False)
    
    print(f"executemany: {tm_many:.4f} seconds")
    print(f"Individual: {tm_single:.4f} seconds")
    print(f"Speedup: {tm_single/tm_many:.1f}x")

该示例创建一个内存数据库，并使用这两种方法对插入 1000 行的时间进行计时。executemany 通常明显更快。

性能优势来自于减少 Python-SQLite 往返和事务开销。 差异随着数据集的增大而增大。

最佳实践

• 用于批量操作： 非常适合插入/更新多行
• 注意内存使用情况： 对于巨大的数据集，请考虑使用生成器
• 正确处理错误： 整个批次会在出错时失败
• 考虑分块： 非常大的批次可能需要拆分
• 使用事务： executemany 在单个事务中运行

资料来源

• Python sqlite3 文档
• SQLite INSERT 文档
• SQLite 性能提示

作者

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

列出所有 Python 教程。