Python sqlite3.Blob.write 方法
上次修改时间：2025 年 4 月 15 日
本综合指南探讨了 Python 的 sqlite3.Blob.write 方法，该方法用于将二进制数据写入 SQLite BLOB 列。我们将涵盖基本用法、参数、实际示例和最佳实践。
基本定义
sqlite3.Blob.write 方法将二进制数据写入 SQLite 数据库中打开的 BLOB（二进制大型对象）。它修改 BLOB 内容在指定偏移量处。
主要特性：它需要打开的 blob 句柄，写入类字节对象，并在事务边界内运行。该方法对于在 SQLite 中高效地操作二进制数据至关重要。
基本 BLOB 写入
此示例演示了创建具有 BLOB 列的表并使用 write 方法写入数据。
basic_write.py
import sqlite3
with sqlite3.connect('blobs.db') as conn:
conn.execute("CREATE TABLE IF NOT EXISTS images (id INTEGER PRIMARY KEY, data BLOB)")
# Insert empty BLOB
conn.execute("INSERT INTO images (data) VALUES (zeroblob(100))")
rowid = conn.execute("SELECT last_insert_rowid()").fetchone()[0]
# Open BLOB for writing
blob = conn.blobopen('images', 'data', rowid)
with blob:
# Write binary data at offset 0
blob.write(b'\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR')
# Verify
data = conn.execute("SELECT substr(data, 1, 16) FROM images WHERE id = ?", (rowid,)).fetchone()[0]
print(f"First 16 bytes: {data}")
此示例创建一个表，插入一个 100 字节的空 BLOB，然后将 PNG 标头写入其中。blobopen 方法返回一个 Blob 对象。
write 方法在当前位置（默认偏移量为 0）写入字节。始终在 with 语句中使用 Blob 对象以进行适当的资源管理。
在特定偏移量写入
write 方法可以在 BLOB 中的任何偏移量处写入数据。此示例显示在非零偏移量处写入。
offset_write.py
import sqlite3
with sqlite3.connect('blobs.db') as conn:
conn.execute("CREATE TABLE IF NOT EXISTS documents (id INTEGER PRIMARY KEY, content BLOB)")
conn.execute("INSERT INTO documents (content) VALUES (zeroblob(500))")
rowid = conn.execute("SELECT last_insert_rowid()").fetchone()[0]
blob = conn.blobopen('documents', 'content', rowid)
with blob:
# Move to offset 100
blob.seek(100)
# Write data at offset 100
blob.write(b'CHAPTER 1: Introduction')
# Verify
excerpt = conn.execute("SELECT substr(content, 100, 22) FROM documents WHERE id = ?", (rowid,)).fetchone()[0]
print(f"Excerpt: {excerpt}")
此示例从 500 字节 BLOB 中的偏移量 100 开始写入文本数据。seek 方法在写入之前定位写入光标。
在特定偏移量处写入对于更新大型二进制对象的各个部分而无需重写整个内容非常有用。
追加到 BLOB
此示例演示了如何通过在结束位置写入将数据追加到现有 BLOB。
append_write.py
import sqlite3
with sqlite3.connect('blobs.db') as conn:
conn.execute("CREATE TABLE IF NOT EXISTS logs (id INTEGER PRIMARY KEY, entries BLOB)")
conn.execute("INSERT INTO logs (entries) VALUES (?)", (b'LOG START',))
rowid = conn.execute("SELECT last_insert_rowid()").fetchone()[0]
blob = conn.blobopen('logs', 'entries', rowid)
with blob:
# Move to end
blob.seek(0, 2) # 2 means seek relative to end
# Append data
blob.write(b'\nNew log entry at ' + str(int(time.time())).encode())
# Verify
full_log = conn.execute("SELECT entries FROM logs WHERE id = ?", (rowid,)).fetchone()[0]
print(f"Full log:\n{full_log.decode()}")
该示例将带时间戳的日志条目追加到现有 BLOB。具有模式 2 的 seek 在写入之前定位到 BLOB 的末尾。
追加对于存储在数据库中的日志文件或其他顺序增长的二进制数据非常有效。
覆盖部分内容
此示例显示了如何在保留周围数据的情况下覆盖 BLOB 的一部分。
partial_overwrite.py
import sqlite3
with sqlite3.connect('blobs.db') as conn:
conn.execute("CREATE TABLE IF NOT EXISTS config (id INTEGER PRIMARY KEY, settings BLOB)")
initial_data = b'DEFAULT_CONFIG____VERSION_1.0____'
conn.execute("INSERT INTO config (settings) VALUES (?)", (initial_data,))
rowid = conn.execute("SELECT last_insert_rowid()").fetchone()[0]
blob = conn.blobopen('config', 'settings', rowid)
with blob:
# Overwrite version part
blob.seek(20)
blob.write(b'2.0')
# Verify
updated = conn.execute("SELECT settings FROM config WHERE id = ?", (rowid,)).fetchone()[0]
print(f"Updated config: {updated.decode()}")
该示例仅覆盖配置 BLOB 中的版本号，同时保持其余数据不变。seek 定位到版本号位置。
部分覆盖对于更新结构化二进制数据中的特定字段而无需重写所有内容非常有效。
写入大型二进制数据
此示例演示了以块的形式写入大型二进制数据以处理内存约束。
chunked_write.py
import sqlite3
import os
def generate_large_data(size):
"""Generate dummy binary data of specified size"""
return os.urandom(size)
with sqlite3.connect('blobs.db') as conn:
conn.execute("CREATE TABLE IF NOT EXISTS large_files (id INTEGER PRIMARY KEY, content BLOB)")
file_size = 5 * 1024 * 1024 # 5MB
conn.execute("INSERT INTO large_files (content) VALUES (zeroblob(?))", (file_size,))
rowid = conn.execute("SELECT last_insert_rowid()").fetchone()[0]
blob = conn.blobopen('large_files', 'content', rowid)
with blob:
chunk_size = 1024 * 1024 # 1MB chunks
for offset in range(0, file_size, chunk_size):
chunk = generate_large_data(min(chunk_size, file_size - offset))
blob.seek(offset)
blob.write(chunk)
# Verify size
actual_size = conn.execute("SELECT length(content) FROM large_files WHERE id = ?", (rowid,)).fetchone()[0]
print(f"Written {actual_size} bytes")
该示例以 1MB 的块写入 5MB 的数据，以演示处理大型 BLOB。zeroblob 预先分配空间以实现高效写入。
分块写入对于处理大型文件，同时在应用程序中保持合理的内存使用量至关重要。
BLOB 写入中的错误处理
此示例显示了写入 BLOB 时的正确错误处理，包括事务管理。
error_handling.py
import sqlite3
def write_with_retry(conn, table, column, rowid, data, max_attempts=3):
for attempt in range(max_attempts):
try:
blob = conn.blobopen(table, column, rowid)
with blob:
blob.write(data)
return True
except sqlite3.OperationalError as e:
print(f"Attempt {attempt + 1} failed: {e}")
if attempt == max_attempts - 1:
conn.rollback()
return False
continue
return False
with sqlite3.connect('blobs.db') as conn:
conn.execute("CREATE TABLE IF NOT EXISTS critical (id INTEGER PRIMARY KEY, payload BLOB)")
conn.execute("INSERT INTO critical (payload) VALUES (zeroblob(100))")
rowid = conn.execute("SELECT last_insert_rowid()").fetchone()[0]
success = write_with_retry(conn, 'critical', 'payload', rowid, b'IMPORTANT_DATA')
if success:
conn.commit()
print("Write successful")
else:
print("All write attempts failed")
该示例为 BLOB 写入实现了一个重试机制，该机制可能会因数据库锁或其他瞬态问题而失败。每次尝试都包含在适当的错误处理中。
强大的错误处理对于处理关键二进制数据存储的生产应用程序至关重要。
最佳实践
- 使用上下文管理器： 始终在
with语句中打开 Blob - 预分配空间： 对大型 BLOB 使用
zeroblob - 处理错误： 为并发访问实现重试逻辑
- 分块大型写入： 以可管理的部分处理大型 BLOB
- 验证偏移量： 确保写入保持在 BLOB 边界内
资料来源
作者
列出所有 Python 教程。