Python小型JSON存档:先写临时文件,再替换正式文件

举报
yd_243677465 发表于 2026/11/04 20:22:50 2026/11/04
【摘要】 用Python标准库演示小型JSON存档:先序列化、同目录临时写入、flush与fsync、关闭后替换。通过四项本地断言验证正常保存及失败时保留旧文件,并说明原子性、持久性和多进程并发的边界。

今天阅读鸿蒙数独案例时,注意到它把通关进度保存到本地 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 辅助编写,示例已在本机执行验证。

【声明】本内容来自华为云开发者社区博主,不代表华为云及华为云开发者社区的观点和立场。转载时必须标注文章的来源(华为云社区)、文章链接、文章作者等基本信息,否则作者和本社区有权追究责任。如果您发现本社区中有涉嫌抄袭的内容,欢迎发送邮件进行举报,并提供相关证据,一经查实,本社区将立刻删除涉嫌侵权内容,举报邮箱: cloudbbs@huaweicloud.com
  • 点赞
  • 收藏
  • 关注作者

评论(0)

0/1000
抱歉,系统识别当前为高风险访问,暂不支持该操作

全部回复

上滑加载中

设置昵称

在此一键设置昵称,即可参与社区互动!

*长度不超过10个汉字或20个英文字符,设置后3个月内不可修改。

*长度不超过10个汉字或20个英文字符,设置后3个月内不可修改。