Python sqlite3.Connection.blobopen 怎么分块读写 BLOB：事务边界与零拷贝缓冲区

把几百 MB 的备份文件先读成一个 Python bytes，再交给 SQLite，通常会在内存峰值处暴露问题。sqlite3.Connection.blobopen 提供了另一条路：先创建固定大小的 BLOB，再让 Blob 句柄按块写入或读取。关键限制也很明确——句柄不能改变 BLOB 长度，所以空间预留、分块写入和事务提交必须配套设计。

为什么不用一次性 bytes 写入 BLOB

一次性写入的代码很短，但它把文件大小、Python 内存和 SQLite 参数绑定绑在了一起。备份导入、缩略图归档这类任务往往已经有一个文件句柄；为了插入数据库再构造完整 bytes，内存会同时留下输入缓冲和数据库适配器的中间对象。

blobopen 的思路是把“容量”和“内容”拆开：SQL 负责创建一段固定空间，Blob 负责在偏移量上逐块读写。这不是自动扩容的文件系统，读写前必须知道目标大小。

先用 zeroblob 预留空间，再打开 blobopen

下面的示例把一个文件写入 attachments.payload。表需要有 rowid；Python 官方文档明确指出，blobopen 不能用于 WITHOUT ROWID 表。

from pathlib import Path
import sqlite3

CHUNK_SIZE = 64 * 1024

def store_file(con: sqlite3.Connection, path: Path) -> int:
    total = path.stat().st_size
    cur = con.execute(
        "INSERT INTO attachments (name, payload) VALUES (?, zeroblob(?))",
        (path.name, total),
    )
    rowid = cur.lastrowid

    with con.blobopen("attachments", "payload", rowid, readonly=False) as blob:
        with path.open("rb") as source:
            while chunk := source.read(CHUNK_SIZE):
                blob.write(chunk)
    return rowid

这里的真实调用链是 zeroblob → blobopen → 固定长度 → 分块写入：SQL 插入阶段先拿到 rowid，句柄随后定位到这条记录的 BLOB，循环每次只保留一个块。

为什么必须先知道文件大小

Blob.write() 不会把 BLOB 扩大。文件实际长度和预留长度不一致时，要么在写入前截断/分段，要么改用临时文件计算大小；不能指望最后一次 write() 自动扩容。写入超出固定范围会得到 ValueError，这正是应该在测试里覆盖的边界。

Blob 句柄关闭不等于事务已经提交

with con.blobopen(...) 只负责在离开代码块时关闭 Blob 句柄。数据库事务仍由连接控制。生产代码应把“句柄关闭”和“记录提交”分成两个检查点：

def import_file(db: str, source_path: str) -> int:
    con = sqlite3.connect(db)
    try:
        con.execute(
            "CREATE TABLE IF NOT EXISTS attachments "
            "(name TEXT NOT NULL, payload BLOB NOT NULL)"
        )
        rowid = store_file(con, Path(source_path))
        con.commit()
        return rowid
    except Exception:
        con.rollback()
        raise
    finally:
        con.close()

成功路径是 Blob → commit()；失败路径是写入异常 → rollback()。读取时则使用 Blob 的当前位置和 read()，不要把“句柄可读”误认为“数据已经永久保存”。

分块读取、覆盖写和校验

读取端可以继续按块处理，例如把 BLOB 直接写到校验文件，而不是一次性调用 read() 取完全部内容。seek() 还能回到指定偏移，但偏移量必须落在固定 BLOB 范围内。

def export_file(con: sqlite3.Connection, rowid: int, target: Path) -> None:
    with con.blobopen("attachments", "payload", rowid, readonly=True) as blob:
        with target.open("wb") as out:
            while chunk := blob.read(CHUNK_SIZE):
                out.write(chunk)

覆盖写适合修补固定长度的内容，例如替换文件头或更新一段已分配的二进制块；它不适合追加未知大小的数据。若业务需要追加，先在业务层计算新总长度，再创建新的 zeroblob 记录会更稳。

三个容易踩中的边界

还要留意 SQLite 锁：一个长时间保持打开的写句柄会让其他连接更难获得写锁。大文件导入应设置合理的事务范围，写完立即关闭句柄并提交，不要把用户交互或网络下载放在同一个事务里。

常见问题

blobopen 能不能直接创建 BLOB

不能。它打开已有 BLOB；先用 INSERT ... zeroblob(?) 创建固定长度，再传入 rowid。

为什么 Blob.write() 不支持追加

Blob 长度是创建时确定的。追加会改变长度，超出末尾时会抛出 ValueError，应创建新记录或预留更大的空间。

关闭 Blob 后还需要 commit() 吗

需要。关闭句柄只释放 BLOB 资源，写入是否成为持久修改仍取决于连接的事务提交。

读取大 BLOB 是否必须一次性 read()

不必。按块调用 read(CHUNK_SIZE) 能控制内存峰值，只有确实需要完整内容时才一次性读取。

把固定长度当成设计约束

blobopen 适合“大小已知、内容分块到达、需要少占内存”的写入路径。它让文件式读写进入 SQLite，但没有替你解决容量规划和事务边界。先用 zeroblob 把长度定下来，再在 Blob 上做有限范围的读写，最后用 commit() 或 rollback() 明确收口，这三个检查点比单纯追求一段更短的插入代码更重要。