Python sqlite3.blobopen 怎样读写 SQLite 的大二进制字段

如果 SQLite 表里已经有一条 BLOB 记录，Python 不必先把整列读成一个大 bytes 再整体回写。从 Python 3.11 起，sqlite3.Connection.blobopen() 可以打开已有 BLOB，返回一个类似文件的 Blob 句柄，用 read()、write()、seek() 处理局部数据。关键限制是：Blob 句柄只能读写既有容量，不能在句柄里扩容。

官方文档：https://docs.python.org/3/library/sqlite3.html#sqlite3.Connection.blobopen

实际使用时把动作拆成三层：先用 SQL 创建带固定容量的 BLOB，再用表名、列名和 rowid 打开它，最后在 Blob 关闭后提交连接事务。这样既能避免一次性搬运大字段，也能看清失败和资源释放的边界。

blobopen 先解决什么问题

blobopen(table, column, rowid, readonly=False, name='main') 的定位信息不是查询条件，而是一个已经存在的 BLOB 地址。table 是表名，column 是 BLOB 列，rowid 是目标行的整数行号；需要读取时可以把 readonly 设为 True，需要写入时保留默认的可写句柄。

这个 API 不负责创建行，也不负责改变字段类型。若目标行不存在，或者表是 WITHOUT ROWID 表，打开动作就不符合它的定位模型。可以把它理解为“拿到已有 BLOB 的文件句柄”，而不是“用一个新 API 插入 BLOB”。

先用 zeroblob 预留固定容量

因为 Blob 的长度不能改变，写入前要先决定容量。SQLite 提供的 zeroblob(N) 可以在插入记录时预留 N 个零字节。下面只负责创建一条可定位的记录，并保存返回的 rowid：

import sqlite3

con = sqlite3.connect("assets.db")
try:
    # 先准备一张带固定容量 BLOB 的表，rowid 用来定位后续句柄。
    con.execute("""
        CREATE TABLE IF NOT EXISTS media (
            name TEXT NOT NULL,
            payload BLOB NOT NULL
        )
    """)
    cursor = con.execute(
        "INSERT INTO media(name, payload) VALUES(?, zeroblob(?))",
        ("manual.bin", 1024 * 1024),  # 为示例预留 1 MiB，容量由 SQL 决定。
    )
    rowid = cursor.lastrowid
    con.commit()  # 先提交记录，之后 blobopen 才能定位到它。
finally:
    con.close()  # 连接不再使用时显式关闭，避免资源悬挂。

这里的 rowid 只是示例变量，真实项目应把它保存到业务记录或随任务传递。若业务需要更大的文件，先选择合适的固定容量；不要指望后面的 Blob.write() 自动把字段变长。

按偏移读取或写入局部内容

拿到行号后，可以让 blobopen() 返回可写或只读句柄。seek(offset) 移动当前位置，read(length) 读取一段，write(data) 则覆盖当前位置开始的字节。下面的例子把两种方向放在一起，便于比较：

import sqlite3

con = sqlite3.connect("assets.db")
try:
    rowid = 1  # 示例行号；生产代码应使用实际 INSERT 返回的 rowid。

    # 可写句柄只能覆盖已经预留的范围，不能改变 BLOB 总长度。
    with con.blobopen("media", "payload", rowid, readonly=False) as blob:
        blob.seek(128)  # 把写入位置移动到已有 BLOB 的偏移 128。
        blob.write(b"header-v2")  # 写入长度不能超出固定容量。

    # 只读句柄适合读取局部区间，避免把整个大字段一次性载入内存。
    with con.blobopen("media", "payload", rowid, readonly=True) as blob:
        blob.seek(128)  # 读取同一位置，验证需要的局部内容。
        chunk = blob.read(9)

    con.commit()  # 写句柄关闭后再提交连接事务。
finally:
    con.close()  # 连接关闭前，确保上面的 Blob 上下文已经退出。

在这个模型里，readonly=True 是权限意图的明确表达；不要为了统一代码而把所有读取都打开成可写。局部读取适合索引、分片或固定格式记录，但它不会自动替你做并发协调，业务仍需决定什么时候提交和怎样处理写入冲突。

把提交和关闭放在正确边界

Blob 支持上下文管理器，因此优先使用 with con.blobopen(...) as blob。退出上下文后句柄会关闭，后续再访问它会抛出错误。连接本身的事务提交仍应由连接代码负责，尤其是可写句柄退出后再调用 commit()，边界更容易读懂。

import sqlite3

def write_slice(database, rowid, offset, data):
    con = sqlite3.connect(database)
    try:
        # 让异常保留给调用方，调用方可按任务类型决定重试或记录。
        with con.blobopen("media", "payload", rowid, readonly=False) as blob:
            blob.seek(offset)
            blob.write(data)
        con.commit()  # Blob 关闭后提交本次连接事务。
    except (sqlite3.Error, ValueError):
        con.rollback()  # 打开或写入失败时撤销当前连接的未提交变更。
        raise
    finally:
        con.close()  # 无论成功失败都释放连接。

try:
    write_slice("assets.db", 1, 256, b"chunk")
except sqlite3.Error as exc:
    print(f"BLOB 写入失败：{exc}")  # 这里只记录错误，不伪造写入成功。

异常处理的重点是区分“句柄打开失败”和“写入后提交失败”。例如表名、列名或 rowid 不正确时，blobopen() 可能直接抛出 SQLite 异常；如果写入超过容量，也不应把它包装成成功。这里让异常继续向上抛出，调用方才能决定重试、告警或换用新的记录。

版本和容量边界要先确认

如果你的数据长度经常变化，先重新评估表结构和容量策略；blobopen 更适合已有记录、固定容量、需要局部读写的场景。它解决的是大字段搬运方式，不是通用文件系统，也不会替代事务设计。

常见问题

blobopen 能直接打开不存在的 BLOB 吗？

不能。先插入记录并用 zeroblob(N) 预留容量，再用 table、column 和 rowid 打开已有 BLOB。

为什么写入成功后还要 commit？

Blob 句柄属于连接事务。可写句柄关闭只表示句柄资源已释放，连接层仍应按事务策略提交或回滚。

怎样改变 BLOB 的大小？

不要在 Blob 对象上扩容。回到 SQL 层重新写入合适大小的 BLOB，或设计新的固定容量记录。

读取大字段一定要用 blobopen 吗？

不一定。小字段直接查询更简单；当字段较大、只需局部区间或希望分段处理时，Blob 句柄才更有价值。