Python sqlite3.Cursor.fetchmany 方法

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

本综合指南探讨了 Python 的 sqlite3.Cursor.fetchmany 方法，该方法用于从数据库查询结果中检索多行数据。我们将介绍基本用法、参数、内存效率和实际示例。

基本定义

fetchmany 方法从查询结果中检索下一组行。它返回一个元组列表，其中每个元组代表数据库中的一行。

主要特性：它接受一个 size 参数来控制要获取的行数，对于大型结果集具有内存效率，并且在调用之间保持游标位置。它非常适合查询结果的批量处理。

fetchmany 的基本用法

这是 fetchmany 最简单的用法，可以批量检索行。

import sqlite3

with sqlite3.connect('example.db') as conn:
    conn.execute("CREATE TABLE IF NOT EXISTS products (id INTEGER, name TEXT)")
    conn.executemany("INSERT INTO products VALUES (?, ?)", 
                    [(1, 'Laptop'), (2, 'Phone'), (3, 'Tablet'), 
                     (4, 'Monitor'), (5, 'Keyboard')])
    
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM products")
    
    # Fetch first 2 rows
    batch1 = cursor.fetchmany(2)
    print("Batch 1:", batch1)
    
    # Fetch next 2 rows
    batch2 = cursor.fetchmany(2)
    print("Batch 2:", batch2)
    
    # Fetch remaining rows (1 in this case)
    batch3 = cursor.fetchmany(2)
    print("Batch 3:", batch3)

此示例显示了基本批量获取。我们插入 5 行，然后分 2 行一批地获取它们。游标在调用之间保持位置。

输出将显示前两个产品，然后是接下来的两个产品，然后是剩下的一个产品。当没有更多行可用时，该方法返回一个空列表。

使用默认大小进行获取

当未指定大小 (size) 时，fetchmany 使用游标的 arraysize。

import sqlite3

with sqlite3.connect('example.db') as conn:
    cursor = conn.cursor()
    
    # Set the default fetch size
    cursor.arraysize = 3
    cursor.execute("SELECT * FROM products")
    
    # Fetch using default size
    batch = cursor.fetchmany()
    print("First batch (size 3):", batch)
    
    # Change size and fetch again
    cursor.arraysize = 1
    batch = cursor.fetchmany()
    print("Second batch (size 1):", batch)

这演示了当未提供 size 参数时，arraysize 如何影响 fetchmany。第一次获取得到 3 行，第二次获取得到 1 行。

当您希望在整个应用程序中始终使用相同的提取大小时，设置 arraysize 非常有用。

处理大型结果集

fetchmany 非常适合处理大型结果集，而无需将所有内容都加载到内存中。

import sqlite3

def process_large_table():
    with sqlite3.connect('large_db.db') as conn:
        cursor = conn.cursor()
        cursor.execute("SELECT * FROM big_table")
        
        while True:
            batch = cursor.fetchmany(100)  # Process 100 rows at a time
            if not batch:
                break
                
            # Process the batch
            for row in batch:
                process_row(row)

def process_row(row):
    # Simulate row processing
    print(f"Processing row {row[0]}")

# Create sample large database
with sqlite3.connect('large_db.db') as conn:
    conn.execute("CREATE TABLE big_table AS SELECT value as id, 'Data ' || value as info FROM generate_series(1, 1000)")
    
process_large_table()

这种模式对于处理大型数据集至关重要。它以易于管理的块来获取行，并在获取下一批之前处理每个批次。

while 循环会一直持续到 fetchmany 返回一个空列表，表明已处理完所有行。

与 fetchone 结合使用

fetchmany 可以与 fetchone 结合使用，以实现灵活的结果处理。

import sqlite3

with sqlite3.connect('example.db') as conn:
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM products ORDER BY id")
    
    # Get first row individually
    first_row = cursor.fetchone()
    print("First product:", first_row)
    
    # Then get next 2 rows as a batch
    next_rows = cursor.fetchmany(2)
    print("Next two products:", next_rows)
    
    # Then get one more row
    another_row = cursor.fetchone()
    print("Another product:", another_row)

此示例显示了混合提取样式。我们使用 fetchone 获取第一行，然后使用 fetchmany 获取一批，然后再获取单行。

游标在所有提取方法中始终如一地保持位置，从而允许灵活的结果处理。

空结果处理

fetchmany 能够优雅地处理空结果集和部分提取。

import sqlite3

with sqlite3.connect('example.db') as conn:
    cursor = conn.cursor()
    
    # Query with no results
    cursor.execute("SELECT * FROM products WHERE id = 999")
    empty_result = cursor.fetchmany(5)
    print("Empty result:", empty_result)  # []
    
    # Query with fewer results than requested
    cursor.execute("SELECT * FROM products LIMIT 3")
    partial_result = cursor.fetchmany(5)
    print("Partial result (3 rows):", partial_result)
    
    # Subsequent fetch returns empty list
    next_result = cursor.fetchmany(2)
    print("After exhaustion:", next_result)  # []

这演示了 fetchmany 在空或小结果集中的行为。当没有行可用时，它返回一个空列表。

当存在的行数少于请求的行数时，它会返回所有可用行而不会出错。后续调用返回空列表。

与行工厂一起使用

fetchmany 与行工厂一起使用以返回自定义的行格式。

import sqlite3

with sqlite3.connect('example.db') as conn:
    # Use built-in Row factory
    conn.row_factory = sqlite3.Row
    cursor = conn.cursor()
    
    cursor.execute("SELECT * FROM products")
    batch = cursor.fetchmany(2)
    
    # Access rows by column name
    for row in batch:
        print(f"Product {row['id']}: {row['name']}")
        print("Columns:", row.keys())

此示例显示了带有 sqlite3.Row 工厂的 fetchmany。可以通过列名或索引访问行。

keys 方法显示可用的列名，使结果比普通元组更具自描述性。

错误处理

正确的错误处理确保了使用 fetchmany 进行的健壮的数据库操作。

import sqlite3

try:
    with sqlite3.connect('example.db') as conn:
        cursor = conn.cursor()
        
        # Invalid query (no such table)
        cursor.execute("SELECT * FROM non_existent_table")
        batch = cursor.fetchmany(2)
        
except sqlite3.Error as e:
    print("Database error:", e)
    
finally:
    print("Operation attempted")

这显示了处理在提取操作期间可能发生的错误。with 语句确保正确的连接清理。

常见的错误包括无效的 SQL、缺少表或数据库访问错误。始终在生产代码中处理这些情况。

最佳实践

• 使用适当的批量大小： 平衡内存和往返次数
• 始终检查空结果： 处理数据结束的情况
• 与上下文管理器结合使用： 确保资源清理
• 考虑内存效率： 使用 fetchmany 处理大型结果
• 维护游标状态： 注意调用之间的位置

资料来源

• Python sqlite3 文档
• SQLite 游标提取
• SQLite 限制

作者

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

列出所有 Python 教程。