Python sqlite3.Connection.blobopen 怎么增量读写大字段

处理图片、归档包或模型文件时，直接把整个二进制字段读进内存并不划算。Python 3.11 起，标准库 sqlite3 提供 Connection.blobopen()，可以像操作文件一样对数据库中的既有 BLOB 分块读取、写入和定位。

核心结论：先用 zeroblob(size) 创建固定长度 BLOB，再按 rowid 打开 Blob 句柄分块写入；读取时使用 readonly=True。Blob 句柄不能改变字段总长度，需要扩容或缩容时应执行 UPDATE 或重建记录。

官方参考地址：https://docs.python.org/3/library/sqlite3.html

先确认 blobopen 的固定长度边界

blobopen(table, column, rowid, *, readonly=False, name="main") 打开的对象对应某张表、某一列和某一行。它不是创建任意长度文件的接口，而是打开已经存在的 BLOB。调用前要确认四个边界：

• 目标表必须是普通 rowid 表；对 WITHOUT ROWID 表调用会抛出 OperationalError。
• rowid 应来自可信查询或刚插入记录的 lastrowid，不能把外部输入直接当成已授权记录。
• Blob 长度在打开前已经确定，write() 不能越过末尾，也不能通过句柄扩容。
• Blob 是文件式资源，应放进 with 语句，保证成功和异常路径都及时关闭。

先建立最小数据表。元数据和二进制主体分列保存，后续更容易核对文件名、声明长度和写入结果。

import sqlite3

con = sqlite3.connect("files.db")
con.execute("""
    CREATE TABLE IF NOT EXISTS files (
        name TEXT NOT NULL,
        byte_size INTEGER NOT NULL,
        payload BLOB NOT NULL
    )
""")
con.commit()  # 提交建表事务

先分配固定长度，再分块写入

SQLite 的 zeroblob(N) 会生成长度为 N 的零值 BLOB，适合先预留空间，再由增量 I/O 句柄填充。应用层仍要设置明确上限，否则异常请求可能提交极大的声明长度，导致数据库文件和 WAL 快速增长。

图1：SQLite BLOB 增量写入边界，先固定长度，再通过受控句柄分块写入。

CHUNK_SIZE = 1024 * 1024
MAX_BLOB_SIZE = 64 * 1024 * 1024


def save_stream(con, name, source, size):
    # 先限制预分配大小，避免磁盘资源被异常请求耗尽
    if not 0 

占位和填充位于同一事务。若输入提前结束、数据超长或写入异常，事务会回滚，不会留下看似完整、实际只写了一部分的记录。SQL 值全部通过占位符绑定，文件名和长度都不会被拼进 SQL。

如果上游流不允许在末尾额外读取一字节，可以把“声明长度必须准确”的责任前移，并在本层记录实际写入量和摘要。但只要条件允许，最好当场检查，因为 Blob 的固定长度会让未写区域保持零值，单看字段长度无法发现内容不完整。

用只读句柄分块导出

读取无需先查询完整字段。业务查询应先确认当前调用者有权访问目标记录，再把可信 rowid 交给读取函数。打开时声明 readonly=True，把误写风险降到最低。

def export_stream(con, rowid, sink):
    # rowid 应在调用前完成所属关系或访问权限校验
    with con.blobopen(
        "files", "payload", rowid, readonly=True
    ) as blob:
        while True:
            chunk = blob.read(CHUNK_SIZE)
            if not chunk:
                break
            sink.write(chunk)  # 逐块写入文件、响应流或摘要器

内存峰值主要由 CHUNK_SIZE 决定，而不是 BLOB 总大小。块太小会增加 Python 与 SQLite 之间的调用次数；块太大则增加单次内存占用。1 MiB 可作为易懂的起点，生产值应根据存储延迟、并发数和下游背压测试后调整。

连接上下文管理器只负责提交或回滚事务，并不会关闭连接本身。Blob 句柄由内部 with 关闭；连接不再使用时，仍要在外层显式调用 con.close() 或用其他资源管理方式关闭。

把行权限、大小和句柄生命周期守住

增量 I/O 降低了内存压力，但不会自动解决授权和资源滥用。按风险检查如下：

1. 越权 rowid：外部用户提交他人记录编号，应用未做所属关系查询就打开 Blob。先按当前身份查询可访问记录，再使用查询结果。
2. 超大预分配：恶意声明巨大 size，zeroblob 使数据库文件快速增长。应同时设置业务上限、用户配额和磁盘监控。
3. 错误写权限：下载路径使用默认可写句柄。读取函数应明确传入 readonly=True。
4. 句柄泄漏：异常路径未关闭 Blob。始终使用上下文管理器，不把句柄保存到请求作用域之外。
5. SQL 拼接：把名称或条件直接拼进语句。值使用参数化 SQL，表名和列名保持程序内常量。

图2：BLOB 操作的三层防护边界，分别控制访问、资源和审计。

审计日志不应记录二进制正文，可记录调用者、rowid、声明长度、实际字节数、内容摘要、只读或写入模式、耗时和事务结果。摘要能帮助排查传输损坏，也避免把敏感原文复制到日志系统。

替换长度或局部改写时怎么做

若新内容长度和原字段不同，不要尝试通过 Blob 句柄扩容。应在事务中 UPDATE 为新的 zeroblob(new_size)，然后重新打开并完整写入；也可以插入新记录，确认成功后再切换业务引用。后一种方式更容易保留旧版本并做失败恢复。

只修改固定长度字段中的一小段时，可用 seek() 定位，但必须先验证偏移和片段长度没有越界。

def patch_bytes(con, rowid, offset, data):
    with con:  # 局部修改也放在事务中
        with con.blobopen("files", "payload", rowid) as blob:
            # 明确检查边界，避免 write 越过 BLOB 末尾
            if offset  len(blob):
                raise ValueError("修改范围超出 BLOB 长度")
            blob.seek(offset)
            blob.write(data)

同一行被其他语句修改时，已打开的 Blob 句柄可能失效。不要长期缓存句柄，也不要在句柄打开期间对同一记录做无关 UPDATE。授权查询、打开、读写、关闭和提交应形成短而清晰的作用域。

验证清单

• 运行环境是否为 Python 3.11 或更高版本？
• 目标表是否是带 rowid 的普通表，而非 WITHOUT ROWID 表？
• 写入前是否校验 size、用户配额和可用磁盘策略？
• rowid 是否来自授权查询或当前事务的 lastrowid？
• 是否先用 zeroblob(size) 固定长度，再打开 Blob？
• 下载路径是否显式设置 readonly=True？
• Blob 是否总由 with 关闭，连接是否在外层关闭？
• 异常是否回滚，审计是否记录声明长度、实际字节数和结果？

正确标准不只是“文件能写进去”。还应保证内存占用只随块大小增长；未授权 rowid 无法进入操作；超限输入在预分配前被拒绝；短流、长流和越界局部写入都会失败并回滚；成功记录的长度、摘要与导出内容一致。

常见问题

blobopen 能直接创建新 BLOB 吗？

不能。它只打开已有字段。常见做法是先 INSERT 包含 zeroblob(size) 的记录，再用 lastrowid 打开。

Blob 句柄能改变字段长度吗？

不能。需要改变总长度时使用 UPDATE 或新记录替换，然后重新打开句柄写入。

为什么读取要指定 readonly=True？

这是最小权限控制，能降低下载或校验代码误写数据库内容的风险，也让函数意图更明确。

增量读取一定比普通 SELECT 快吗？

不一定。它最明确的收益是避免一次性构造完整 BLOB 对象，降低内存峰值并支持流式处理。吞吐仍取决于块大小、磁盘、事务、并发和下游速度，应使用真实负载测量。