sqlite3.Connection.blobopen 怎样分块读写 BLOB

使用 Python 的 sqlite3.Connection.blobopen() 分块读写 BLOB，要先让目标行包含指定长度的 BLOB，再打开句柄，循环调用 write(chunk) 或 read(chunk_size)。写入前可用 zeroblob(n) 预留容量；句柄只能修改已有区域，不能自动扩容。

我会把它看作一种“固定容量、按块访问”的数据入口。这里以导入本地二进制附件为设计场景，说明我会怎样处理容量、rowid 和事务；示例供读者自行执行，不把未运行的代码描述成真实测量记录。

官方地址：https://docs.python.org/3.14/library/sqlite3.html#sqlite3.Connection.blobopen

先把 BLOB 当成固定长度的字节区域

blobopen("assets", "body", row_id) 指向一行的一个字段，而非一个数据库文件路径。前三个参数分别是表名、列名和整数 rowid；数据库符号名默认是 main。Python 的 Blob 接口从 3.11 加入。

我会使用 id INTEGER PRIMARY KEY 这样的普通 rowid 表来承载附件，将业务定位和字节访问分开。它不适用于 WITHOUT ROWID 表。别把业务字符串编号直接传成 rowid，也别以为调用 blobopen 就会替你插入新记录。

zeroblob(n) 表达 n 字节零值区域。这里 n 应来自可靠的长度信息，并受应用上传配额约束；不能为了方便直接相信外部传入的任意大小。

我会在哪些场景选择分块接口

如果附件较大，或者业务只需要读取某个范围，我更倾向于保留一个有限大小的缓冲区，避免先把完整字段变成 Python bytes。代价是容量要预先确定，句柄和事务也需要更仔细地管理。

对于很小的二进制值，参数化 INSERT 或 UPDATE 往往更直观；如果输入长度未知、需要不断追加，固定长度 BLOB 就未必是最合适的模型。不能仅凭“使用了分块接口”就声称数据库吞吐一定更高。

一个有完整收尾的读写示例

准备 Python 3.11 及以上环境和一份稳定的 sample.bin。下面的函数在独立连接上管理事务，写入新行；若数据库已有同名表，需确保其结构与示例一致。不要直接对不熟悉的业务库运行示例。

import sqlite3
from pathlib import Path

CHUNK_SIZE = 64 * 1024

def store_file(source_path, database_path):
    # 输入应是导入期间不会变化的本地普通文件
    with Path(source_path).open("rb") as source:
        source.seek(0, 2)
        expected = source.tell()
        source.seek(0)
        # 显式 SQL 管理事务，兼容 Python 3.11 的连接参数
        con = sqlite3.connect(database_path, isolation_level=None)
        try:
            con.execute("""CREATE TABLE IF NOT EXISTS assets (
                id INTEGER PRIMARY KEY,
                body BLOB NOT NULL
            )""")
            con.execute("BEGIN")
            cursor = con.execute(
                "INSERT INTO assets(body) VALUES(zeroblob(?))", (expected,))
            row_id = cursor.lastrowid
            cursor.close()
            written = 0
            # 关闭句柄之后再提交，异常不能留下半成品记录
            with con.blobopen("assets", "body", row_id) as blob:
                while True:
                    chunk = source.read(CHUNK_SIZE)
                    if not chunk:
                        break
                    if written + len(chunk) > len(blob):
                        raise ValueError("源文件比预留容量更大")
                    blob.write(chunk)
                    written += len(chunk)
                if written != expected:
                    raise ValueError("源文件长度发生变化")
            con.execute("COMMIT")
            return row_id
        except BaseException:
            # 取消或读写错误都回滚当前事务，再向调用者报告
            if con.in_transaction:
                con.execute("ROLLBACK")
            raise
        finally:
            con.close()

def count_blob_bytes(database_path, row_id):
    # 按块消费数据；实际业务可在循环里计算摘要或解析二进制
    con = sqlite3.connect(database_path, isolation_level=None)
    try:
        total = 0
        with con.blobopen("assets", "body", row_id, readonly=True) as blob:
            while True:
                chunk = blob.read(CHUNK_SIZE)
                if not chunk:
                    break
                total += len(chunk)
        return total
    finally:
        con.close()

# 示例调用会创建本地数据库，路径应由调用方明确提供
if __name__ == "__main__":
    row_id = store_file("sample.bin", "assets-demo.db")
    print("row_id:", row_id)
    print("bytes_read:", count_blob_bytes("assets-demo.db", row_id))

缓冲区选 64 KiB 只是示例取值，不是性能结论。写入时记录实际字节数，并将它与预留容量比较；读取时将每块立即消费，避免再把所有块追加到一个列表中。代码中的 bytes_read 只报告消费量，不是内容一致性证明。

源文件若在导入期间被修改，即便长度不变，也可能得到不一致的内容。因此这个示例明确要求稳定输入；严格场景应先准备不可变快照或受应用控制的暂存文件。长度比较只能发现部分变化，不能代替这项约束。

不能把分块写入当成追加文件

偏移量的单位是字节。若只想改固定范围，可以先 seek(offset) 再写入；长度仍然必须合适。若业务需要调整整体长度，应在关闭旧句柄之后用 SQL 重新安排字段容量，而不是在旧句柄上期待“追加成功”。

SQLite 还限制某些带索引或约束的列进行增量写入，所以示例把 body 保持为普通 BLOB 列。设计时应核对表结构，不把所有打开失败都归因于 Python 版本。

资源与事务分别承担什么责任

我更在意的是两个边界：with blob 负责释放句柄，显式事务负责这一批数据库修改是否保留。连接仍需独立关闭。关闭句柄本身不能替代应用层“导入完整才提交”的决定。

示例先关闭写句柄再执行 COMMIT；出错时回滚新插入的行。实际 SQLite 数据库同一时刻只有一个写事务，长时间导入会影响其他写入者，因此我不会在持有写事务时等待交互输入或远程下载。

打开句柄期间，还应避免对同一行执行其他 UPDATE、DELETE 或产生相应冲突副作用。SQLite 可能把该句柄标记为失效；失效不是“之前所有写入已自动回滚”。这也是我把异常交还给外层事务决定的原因。

选择清单与常见问题

• 长度已知且需要局部或分块访问：考虑 blobopen。
• 字段很小：优先选择简单的参数化 SQL。
• 长度未知或频繁追加：先重新评估存储模型与暂存方案。
• 存在并发写入：明确事务时长、失败处理和输入稳定性。

为什么不能像文件一样一直 write？

打开的是已存在、长度固定的数据库值。写入只改变其中的内容，不能让它自动增长。

分块 read 后再 join 是否还节省内存？

如果把所有块都保留下来再拼接，应用仍会持有完整内容。分块的收益依赖消费者及时处理并释放每块。

读取数量相同，是否说明数据完整一致？

不能。字节数相同不代表内容相同；有完整性要求时，应在稳定输入基础上定义摘要或业务格式检查。

对我来说，这个接口的价值是把“行中固定字节区域”暴露成可定位的对象。选型时同时接受固定容量、事务占用和句柄生命周期，分块读写才有清楚的适用范围。

参考资料：Python sqlite3 接口；SQLite 增量 BLOB 打开规则；固定长度写入规则；SQLite 事务；zeroblob 函数。