Python sqlite3.Blob.read 方法

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

本综合指南探讨了 Python 的 sqlite3.Blob.read 方法，该方法用于从 SQLite BLOB 列中读取二进制数据。我们将介绍基本用法、参数和实际示例，并进行适当的资源管理。

基本定义

sqlite3.Blob 类表示 SQLite 数据库中的二进制大对象 (BLOB)。它提供了有效读取和写入二进制数据的方法。

read 方法从 BLOB 列中检索二进制数据。它接受参数来控制读取多少数据以及从何处开始读取。

基本 BLOB 读取

本示例演示了如何使用 read 方法，通过适当的资源管理，从数据库中读取整个 BLOB。

import sqlite3

# Create a database with BLOB data
with sqlite3.connect('blobs.db') as conn:
    conn.execute("CREATE TABLE IF NOT EXISTS images (id INTEGER PRIMARY KEY, data BLOB)")
    
    # Insert sample binary data
    with open('sample.png', 'rb') as f:
        conn.execute("INSERT INTO images (data) VALUES (?)", (f.read(),))
    
    # Read the BLOB
    with conn.blobopen('images', 'data', 1, 'main') as blob:
        image_data = blob.read()
        print(f"Read {len(image_data)} bytes from BLOB")

print("Database connection automatically closed")

本示例创建一个包含 BLOB 列的数据库，从文件中插入二进制数据，然后使用 blob.read 将其读回。 with 语句确保适当的资源清理。

blobopen 方法为指定的表、列和行创建一个 Blob 对象。当不带参数调用时，read 方法读取所有数据。

读取部分 BLOB 数据

read 方法可以通过指定大小和偏移量参数来读取 BLOB 的特定部分。

import sqlite3

with sqlite3.connect('partial.db') as conn:
    conn.execute("CREATE TABLE IF NOT EXISTS documents (id INTEGER, content BLOB)")
    
    # Insert large binary data
    large_data = b'X' * 1024 * 1024  # 1MB of data
    conn.execute("INSERT INTO documents VALUES (1, ?)", (large_data,))
    
    # Read first 100 bytes
    with conn.blobopen('documents', 'content', 1, 'main') as blob:
        first_chunk = blob.read(100)
        print(f"First 100 bytes: {len(first_chunk)}")
        
        # Read next 200 bytes from offset 100
        next_chunk = blob.read(200, 100)
        print(f"Next 200 bytes: {len(next_chunk)}")

print("Resources automatically released")

本示例演示了读取大型 BLOB 的特定部分。第一个 read 获取前 100 个字节，而第二个从偏移量 100 开始读取 200 个字节。

部分读取对于处理大型 BLOB 非常有用，而无需一次将所有内容加载到内存中，从而提高内存效率。

分块读取 BLOB

对于非常大的 BLOB，您可以使用带有 read 方法的循环分块读取数据。

import sqlite3

CHUNK_SIZE = 4096  # 4KB chunks

with sqlite3.connect('large_blobs.db') as conn:
    conn.execute("CREATE TABLE IF NOT EXISTS videos (id INTEGER, frames BLOB)")
    
    # Insert sample video data (simulated)
    video_data = b'FRAME' * 100000  # ~500KB
    conn.execute("INSERT INTO videos VALUES (1, ?)", (video_data,))
    
    # Process in chunks
    with conn.blobopen('videos', 'frames', 1, 'main') as blob:
        total_read = 0
        while True:
            chunk = blob.read(CHUNK_SIZE, total_read)
            if not chunk:
                break
            total_read += len(chunk)
            print(f"Processed chunk {len(chunk)} bytes, total {total_read}")

print(f"Finished processing {total_read} bytes")

本示例以 4KB 的块读取一个大的 BLOB，分别处理每个块。循环持续到 read 返回一个空字节对象为止。

分块读取对于处理非常大的二进制对象，同时保持应用程序中合理的内存使用至关重要。

使用不同偏移量读取 BLOB

read 方法的 offset 参数允许从 BLOB 中的任何位置读取，从而可以随机访问二进制数据。

import sqlite3

with sqlite3.connect('offset.db') as conn:
    conn.execute("CREATE TABLE IF NOT EXISTS binary (id INTEGER, data BLOB)")
    
    # Insert patterned data
    data = bytes(range(256))  # 0-255 byte values
    conn.execute("INSERT INTO binary VALUES (1, ?)", (data,))
    
    # Read from various offsets
    with conn.blobopen('binary', 'data', 1, 'main') as blob:
        # Read first 10 bytes
        print("Start:", blob.read(10))
        
        # Read 10 bytes from middle
        print("Middle:", blob.read(10, 128))
        
        # Read last 10 bytes
        print("End:", blob.read(10, 246))

print("All reads completed")

本示例插入一个包含字节 0-255 的 BLOB，然后演示从不同位置读取。 每个 read 操作都指定大小和偏移量。

随机访问对于具有已知结构的格式特别有用，在这种情况下，您需要读取二进制数据的特定部分而无需处理所有内容。

处理空 BLOB

read 方法对空 BLOB 的行为不同，返回一个空字节对象。 本示例演示了正确的处理方法。

import sqlite3

with sqlite3.connect('empty.db') as conn:
    conn.execute("CREATE TABLE IF NOT EXISTS files (name TEXT, content BLOB)")
    
    # Insert empty BLOB
    conn.execute("INSERT INTO files VALUES ('empty.txt', ?)", (b'',))
    
    # Attempt to read
    with conn.blobopen('files', 'content', 1, 'main') as blob:
        data = blob.read()
        if not data:
            print("BLOB is empty")
        else:
            print(f"Read {len(data)} bytes")

print("Database connection closed")

本示例插入一个空 BLOB（零长度字节），并演示如何检查此情况。 对于空 BLOB，read 方法返回 b''。

处理可能为空的 BLOB 时，始终检查 read 的返回值，以避免将空数据作为有效内容处理。

BLOB 读取的错误处理

本示例演示了读取 BLOB 时的正确错误处理，包括 BLOB 不存在或无效的情况。

import sqlite3
from contextlib import suppress

with sqlite3.connect('errors.db') as conn:
    conn.execute("CREATE TABLE IF NOT EXISTS resources (id INTEGER, payload BLOB)")
    
    try:
        # Attempt to read non-existent BLOB
        with conn.blobopen('resources', 'payload', 999, 'main') as blob:
            data = blob.read()
            print(f"Read {len(data)} bytes")
    except sqlite3.OperationalError as e:
        print(f"Error reading BLOB: {e}")
    
    # Insert invalid data (simulate corruption)
    conn.execute("INSERT INTO resources VALUES (1, 'not a blob')")
    
    with suppress(sqlite3.OperationalError):
        with conn.blobopen('resources', 'payload', 1, 'main') as blob:
            data = blob.read()
            print("This won't print for invalid BLOB")

print("Cleanup complete")

本示例显示了两种错误情况：读取不存在的行以及尝试读取无效的 BLOB 数据。 第一个使用 try/except，第二个使用 suppress。

处理 BLOB 时，正确的错误处理至关重要，因为无效数据或丢失的行可能会导致需要捕获和妥善处理的异常。

最佳实践

• 始终使用上下文管理器： 确保适当的资源清理
• 对于大型 BLOB，请分块读取： 避免内存问题
• 检查空 BLOB： 正确处理零长度数据
• 验证偏移量： 防止读取超出 BLOB 边界
• 妥善处理错误： 捕获潜在的 OperationalError

资料来源

• Python sqlite3 文档
• SQLite BLOB 处理
• SQLite 文件格式

作者

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

列出所有 Python 教程。