Python小型JSON存档:先写临时文件,再替换正式文件
今天阅读鸿蒙数独案例时,注意到它把通关进度保存到本地 JSON。这里由此做一个独立的 Python 小实验:如果保存过程中出错,怎样尽量保住上一份可读存档?本文不修改案例源码,也不声称已经运行鸿蒙应用;下方代码只在本机临时目录内验证文件保存流程。
直接覆盖为什么容易留下半份文件
假设 progress.json 已经记录了通关列表。用 open(path, “w”) 打开后,原文件会先被截断;如果后续序列化失败、写入失败或进程中途退出,原来的完整内容可能已不在。即使 JSON 只有几行,也值得把“生成新内容”和“替换旧文件”分成两步。
本例先把对象序列化为字符串,再在目标文件的同一个目录创建随机临时文件。等临时文件写入、刷新并关闭以后,才调用 os.replace。失败时把异常交还调用方,普通异常路径下清理本次创建的临时文件。目标目录需要事先存在,调用方也应具有写入、替换和删除权限。
可直接运行的标准库示例
把下面代码保存为 atomic_json_demo.py,然后执行 python atomic_json_demo.py。代码只使用标准库,演示数据完全自建,测试目录结束后由 TemporaryDirectory 回收。
"""Small single-writer JSON save demo; does not claim crash durability."""
import json
import os
import tempfile
from pathlib import Path
def save_json(path, value):
path = Path(path)
# Serialize first: invalid data must not truncate the previous save.
payload = json.dumps(value, ensure_ascii=False, allow_nan=False, indent=2)
temporary = None
try:
with tempfile.NamedTemporaryFile(
mode="w", encoding="utf-8", newline="\n",
dir=path.parent, prefix=".save-", suffix=".tmp", delete=False
) as handle:
temporary = Path(handle.name)
handle.write(payload + "\n")
handle.flush()
os.fsync(handle.fileno())
os.replace(temporary, path)
finally:
if temporary is not None:
temporary.unlink(missing_ok=True)
if __name__ == "__main__":
from unittest.mock import patch
with tempfile.TemporaryDirectory() as directory:
target = Path(directory) / "progress.json"
save_json(target, {"solved_ids": ["easy-1"], "note": "已完成"})
assert json.loads(target.read_text(encoding="utf-8"))["note"] == "已完成"
print("PASS: UTF-8 round trip")
old = target.read_bytes()
try:
save_json(target, {"score": float("nan")})
except ValueError:
pass
else:
raise AssertionError("NaN was accepted")
assert target.read_bytes() == old
print("PASS: invalid value preserves previous save")
with patch("os.replace", side_effect=PermissionError("simulated busy file")):
try:
save_json(target, {"solved_ids": ["easy-1", "easy-2"]})
except PermissionError:
pass
else:
raise AssertionError("replace failure was not propagated")
assert target.read_bytes() == old
assert list(Path(directory).glob(".save-*.tmp")) == []
print("PASS: replace failure preserves save and cleans temporary file")
save_json(target, {"solved_ids": ["easy-1", "easy-2"]})
assert len(json.loads(target.read_text(encoding="utf-8"))["solved_ids"]) == 2
print("PASS: successful replacement")
本次实际验证了什么
本机运行输出如下,四项断言全部通过:
PASS: UTF-8 round trip
PASS: invalid value preserves previous save
PASS: replace failure preserves save and cleans temporary file
PASS: successful replacement
第一项验证中文内容能写入并读回。第二项故意传入 NaN;allow_nan=False 会在创建临时文件前拒绝这个值,旧文件的字节保持不变。第三项用 unittest.mock 模拟替换阶段抛出 PermissionError,确认异常没有被悄悄吞掉、旧文件仍可读,并且没有留下本次临时文件。第四项恢复正常替换,把列表从一条更新为两条。
模拟 PermissionError 只验证程序的异常分支,并不代表已经复现某个具体操作系统的文件占用问题,也不能替代断电测试。
几个容易混淆的边界
原子替换解决的是正式路径切换时避免暴露半份新内容。Python 文档说明 os.replace 成功时重命名是原子操作,跨文件系统则可能失败,因此这里把临时文件放在目标目录。(参考:Python官方文档,os模块,os.replace条目)
写入缓冲区与落盘请求也要区分。代码先 flush,再 fsync,然后关闭文件并替换;这遵循了官方对有缓冲文件对象的调用顺序。(参考:Python官方文档,os模块,os.fsync条目)
但这段代码没有同步替换后的目录元数据,也没有验证磁盘、文件系统、网络盘和硬件的掉电行为,不能据此保证断电后最新一次更新一定存在。finally 同样无法在强制终止进程或断电时保证执行,异常退出可能遗留临时文件;后续清理必须先确认归属与使用状态。
另外,原子替换不等于多写入者并发安全。两个进程各自读取旧存档、分别追加一条记录,再顺序替换,仍可能后写覆盖先写。需要并发更新时,应采用单写入队列、适当的锁或数据库事务。此示例适用于受信任本地目录中的小文件、单写入者场景;新文件权限也可能与原文件不同,不能直接当作保留权限、ACL或符号链接语义的通用保存工具。
为什么临时文件要先关闭
NamedTemporaryFile 使用 delete=False,让文件关闭后仍保留名称,之后交给 os.replace;finally 只清理该次调用持有的临时路径。先关闭句柄也避免在 Windows 上依赖仍打开文件的共享删除行为。(参考:Python官方文档,tempfile模块,NamedTemporaryFile条目)
读取端则应区分文件不存在与 JSON 损坏:首次运行没有存档可以初始化;发现现有存档无法解析时,应报告并保留原文件,避免把错误当成“空进度”再自动覆盖。保存成功只说明写入流程完成,应用还需要校验业务字段、版本号和重复记录。
参考阅读:华为云鸿蒙开发实践中心《数独小游戏:HarmonyOS + ArkUI 纯端侧开发》。本文的 Python 保存函数、模拟测试与说明为独立示例,未复制案例实现;代码与文稿由 AI 辅助编写,示例已在本机执行验证。
- 点赞
- 收藏
- 关注作者
评论(0)