结构化数据、序列化、压缩与 SQLite
本节目标
查询 Python JSON、CSV、TOML、pickle、压缩归档和 SQLite 数据边界。
数据格式先确定是否跨语言交换、数据是否可信、字节布局和资源归属，再决定是否需要压缩或查询。以下以 json、csv、tomllib、pickle、struct、gzip、zipfile、tarfile 和 sqlite3 的 Python 3.14 文档为准。文本与字节的入口边界见文本处理、正则表达式与 Unicode，内存中聚合和遍历数据可接续函数式、集合与迭代工具。
JSON 数据模型与定制
JSON 对象、数组、字符串、数值、布尔值和 null 只覆盖一部分 Python 值；json.dumps() 应输出约定的交换格式，而不是任意对象的无损档案。跨系统的金额、大整数和小数精度必须另立合同，因为 JSON 数字没有 Python int 与 float 的全部精度和类型区分；日期、枚举和精确数值也应显式转换为约定的 JSON 基本类型。
import json
payload = json.dumps({"name": "Ada", "score": 95}, separators=(",", ":"))
default= 或自定义编码器只能定义发送方如何变换对象，接收方仍要验证字段、范围和版本；loads() 的结果也不自动恢复领域类型。不要把 JSON 当作签名、模式验证或无限精度的替代，更不要让不受控对象的 __dict__ 成为长期接口。
CSV dialect、换行与字段
CSV 是带 dialect 的无类型表格文本，分隔符、引号、转义和行结束规则都是输入输出合同；字段需要由应用转换成数值、日期或空值。读写时使用已声明的 dialect，并以 newline="" 打开 CSV 文件，让换行由 csv 模块而不是文本层统一处理。
import csv
from io import StringIO
rows = list(csv.reader(StringIO("name,score\r\nAda,95\r\n"), dialect="excel"))
split(",") 不理解引号内逗号，csv.writer 会把 None 写成空字符串而不会保存原始类型。遗漏 newline="" 可能产生额外空行或平台相关的换行处理；未知 dialect、字段数和字段大小也应在读取前限制或验证。
TOML 读取与配置边界
TOML 适合声明式配置读取，解析结果是 Python 基本容器和值，但配置文件的字段仍要按应用规则校验。tomllib.load() 要求二进制输入流，标准库没有 TOML writer；因此输入应以二进制模式交给 tomllib，格式解析则与配置授权、默认值和演进策略分开处理。
import tomllib
from io import BytesIO
config = tomllib.load(BytesIO(b"[server]\nport = 8000\n"))
把文本流传给 tomllib.load() 会违反其输入 API；自行拼接字符串“写回 TOML”也不能保证转义、类型或注释语义。配置可被解析不等于配置值可被信任，文件来源、路径和允许的键仍是调用方的责任。
pickle 的信任与版本边界
pickle 保存 Python 对象图并可支持引用关系和自定义还原，但它是 Python 专用的执行型序列化协议，不是跨语言交换格式。载入范围只能限于同一受控信任域内由自己保存的数据；长期互操作数据应改用有明确模式的格式。
import pickle
blob = pickle.dumps({"name": "Ada"})
不可信的 pickle 载入时可以执行任意代码，因此哈希、扩展名或“看起来像本地文件”都不构成安全许可。pickle 也不承诺跨 Python 版本、类路径或依赖版本的永久兼容；不要对网络输入或用户上传内容调用 pickle.loads()。
struct 与二进制布局
struct 在 Python 值和固定宽度字节布局之间转换，字节序、符号、宽度与对齐必须由协议显式指定。网络或磁盘协议应写出 >、<、! 等明确格式前缀，并在解包前确认缓冲区长度满足布局要求。
import struct
packet = struct.pack(">H", 513)
原生模式可能引入本机字节序和对齐填充，不能用作跨平台格式的默认假设。unpack() 不会替你辨别消息类型、完整性或版本；把任意长度字节解包，或忽略字段范围和协议演进，会导致解析错误或错误解释。
gzip、bz2 与 lzma 压缩流
压缩只缩小字节流，不定义其中数据的结构、可信度或加密性；gzip、bz2 和 lzma 的压缩率、速度和互操作生态也不同。数据应先序列化为明确的字节格式，再按传输约束选择压缩器；若流式 gzip 输出需要可重复，还要显式固定 GzipFile 元数据中的 mtime。
import gzip
compressed = gzip.compress(b"Python", mtime=0)
Python 3.14 中，gzip.compress() 的 mtime 默认为 0；显式传入 None 才使用当前时间。GzipFile 在省略 mtime 或传入 None 时使用当前时间；gzip.open() 写入走该流式行为且不暴露 mtime 参数。压缩不会防篡改或限制解压后的膨胀大小；不要把压缩结果当作保密、签名或输入验证，并对不可信压缩数据设置读取和解压资源限制。
ZIP、TAR 与解包安全
ZIP 和 TAR 归档保存成员名、元数据，且可能携带目录或链接；提取操作会把这些声明映射到本地文件系统。不可信归档若直接 extractall()，其中的路径和链接就会越过调用方审查，因此必须先逐项在受控根目录下验证，再按允许类型提取。
from pathlib import PurePosixPath
member = PurePosixPath("notes/lesson.txt")
归档成员路径、链接目标和解包后的文件数、总大小等资源上限都必须先验证，以防路径穿越、链接逃逸和解压炸弹。仅清理 .. 字符串不能处理绝对路径、平台分隔符或链接语义；即使成员名通过检查，提取期间也要保持根目录与资源所有权边界。
SQLite 连接与事务
SQLite 连接拥有数据库会话与事务状态；with connection: 本身不启动事务，块内操作若已打开事务，退出时才根据结果提交或回滚，而连接生命周期仍由创建者负责。必须同时成功的一组写入应放进连接上下文，连接则无论事务结果如何都由创建者在最终路径显式关闭。
connection = sqlite3.connect(":memory:")
try:
with connection:
connection.execute("INSERT INTO scores VALUES (?, ?)", ("Ada", 95))
finally:
connection.close()
with connection: 只在退出时提交或回滚已经打开的事务；它不会打开新事务，也不会关闭连接。退出时若没有打开的事务，或 autocommit=True，上下文管理器不做任何事。未调用 close() 会让连接及其资源所有权模糊；不要把一次 execute() 的成功视为已持久化，也不要将长事务跨越用户输入、网络等待或不相关工作。
import gzip
import json
import sqlite3
connection = sqlite3.connect(":memory:")
try:
connection.execute("CREATE TABLE scores (name TEXT PRIMARY KEY, score INTEGER NOT NULL)")
with connection:
connection.executemany(
"INSERT INTO scores (name, score) VALUES (?, ?)",
[("Lin", 91), ("Ada", 95), ("Sam", 80)],
)
rows = connection.execute(
"SELECT name, score FROM scores WHERE score >= ? ORDER BY score DESC",
(90,),
).fetchall()
finally:
connection.close()
payload = json.dumps(
[{"name": name, "score": score} for name, score in rows],
ensure_ascii=False,
separators=(",", ":"),
)
compressed = gzip.compress(payload.encode("utf-8"), mtime=0)
roundtrip = gzip.decompress(compressed).decode("utf-8") == payload
print(f"rows={rows}")
print(f"json={payload}")
print("transaction=committed")
print(f"compressed-roundtrip={roundtrip}")
print("parameterized=True")
rows=[('Ada', 95), ('Lin', 91)]
json=[{"name":"Ada","score":95},{"name":"Lin","score":91}]
transaction=committed
compressed-roundtrip=True
parameterized=True
报告只使用内存数据库：事务写入后按分数显式排序，紧凑 JSON 再以 mtime=0 gzip 压缩并解压核对。它证明该固定输入的往返与事务边界，不代表文件数据库并发、迁移或损坏恢复策略。
参数绑定、游标与行工厂
SQL 占位符把数据值交给数据库绑定，游标和行工厂决定如何消费结果；查询若需要稳定展示或测试结果，必须在 SQL 中写出 ORDER BY。所有外部值都通过 ? 或命名占位符绑定；有限的标识符从白名单选择，行访问则明确约定 tuple 或 sqlite3.Row。
rows = connection.execute(
"SELECT name, score FROM scores WHERE score >= ? ORDER BY score DESC",
(90,),
).fetchall()
占位符只能绑定值，不能绑定表名、列名或 ORDER BY 方向等 SQL 标识符；那些结构必须来自固定 SQL 或严格白名单。没有 ORDER BY 的结果顺序不是接口承诺，且延迟消费游标时连接关闭或事务变化会改变其资源和可见性边界。
格式选择与数据完整性
JSON、CSV、TOML、pickle、二进制布局、压缩归档和 SQLite 分别解决交换、表格、配置、受控 Python 对象、协议、打包和关系查询等不同问题。格式选择应同时衡量互操作、信任、精度、体积、查询、版本演进和资源限制；校验、认证与备份仍是独立的完整性措施，不能由格式代劳。
format_name = "json" if needs_interoperability else "sqlite"
格式能解析、压缩能往返或数据库能提交，都不证明来源可信、字段完整或历史版本可读；哈希也只能说明选定字节的完整性，不能替代来源认证。不要用 pickle 处理不可信交换数据、用 CSV 承担嵌套类型，或把归档和 SQLite 当作自动完成备份、访问控制与迁移的系统。