用 Python 原子写入安全保存 JSON 文件

批处理脚本完成了一轮计算，打开 progress.json 准备保存进度，却在写到一半时退出。下次启动时，脚本读到的是空文件或残缺内容，原来的检查点也没了。

对于小型 JSON 状态文件，可以先在目标文件旁边写好新文件，成功后再替换目标。本文用 Python 标准库实现这个流程，并检查保存失败时旧内容是否保留。

1. 确定保存的边界

这个方法适合由单个进程维护的小型进度文件、生成的配置或实验摘要。它可以配合数据集校验清单使用：校验清单确定输入文件，进度文件记录任务运行到哪里。

保存顺序如下：

serialize JSON → write sibling temporary file → flush → fsync → close → replace

Python 文档说明，成功的 os.replace 重命名操作具有原子性，这是 POSIX 的要求。把两个文件放在同一目录，可避免跨文件系统移动。在支持相应语义的本地文件系统上，新打开目标路径的读者会读到完整的旧文件或新文件；已经打开旧文件的读者仍可能继续读取旧内容。

本文假定目录可信、目标是普通文件路径，且只有一个写入者。这个方法不提供锁，也不能把多个文件合并成一笔事务。用于网络存储前，需要验证实际文件系统的行为。

2. 下载并运行示例

使用 Python 3.10 或更新版本，无需安装第三方包。示例已在 macOS 的 Python 3.14.7 上测试。

下载 atomic_json.py，保存到工作目录，然后执行：

python3 --version
python3 atomic_json.py --help
mkdir atomic-json-demo
python3 atomic_json.py atomic-json-demo/progress.json
python3 -m json.tool atomic-json-demo/progress.json

演示程序会替换指定的输出文件，因此请使用新建的演示目录，不要传入已有检查点。前两行输出为：

Rejected NaN; previous file unchanged
OK: completed_batches=2

最终文件内容为：

{
  "completed_batches": 2,
  "status": "ready"
}

这些数值只用于演示保存行为。程序不会执行或恢复训练任务。

3. 完整实现

"""Save a small JSON document by replacement. Python 3.10+, single writer."""
import argparse
import json
import os
from pathlib import Path
import tempfile


def atomic_write_json(path, data):
    """Replace a file in a trusted local directory; parent must already exist."""
    target = Path(path).absolute()
    # Serialize first: invalid data must not touch the destination.
    text = json.dumps(data, ensure_ascii=False, allow_nan=False, indent=2) + "\n"
    temporary = None
    try:
        with tempfile.NamedTemporaryFile(
            mode="w", encoding="utf-8", newline="\n",
            dir=target.parent, prefix=f".{target.name}.",
            suffix=".tmp", delete=False,
        ) as handle:
            temporary = Path(handle.name)
            handle.write(text)
            handle.flush()
            os.fsync(handle.fileno())
        # Close before replacement, including on Windows.
        os.replace(temporary, target)
    finally:
        if temporary is not None:
            temporary.unlink(missing_ok=True)


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("output", type=Path, help="demo JSON path; replaced if present")
    args = parser.parse_args()
    old = {"completed_batches": 1, "status": "ready"}
    atomic_write_json(args.output, old)
    before = args.output.read_bytes()
    try:
        atomic_write_json(args.output, {"loss": float("nan")})
    except ValueError:
        assert args.output.read_bytes() == before
        print("Rejected NaN; previous file unchanged")
    new = {"completed_batches": 2, "status": "ready"}
    atomic_write_json(args.output, new)
    assert json.loads(args.output.read_text(encoding="utf-8")) == new
    print("OK: completed_batches=2")


if __name__ == "__main__":
    main()

序列化发生在创建文件之前。json.dumps 返回字符串；allow_nan=False 会拒绝非有限浮点数，避免输出不符合 JSON 标准的 NaN 和 Infinity。这只检查数据能否编码为 JSON，不会验证业务字段。批次编号等字段应在调用函数之前检查。

NamedTemporaryFile 提供唯一文件名。delete=False 使文件关闭后仍然保留，dir=target.parent 把它放到目标文件旁边。普通异常发生后，finally 会清理临时文件；进程被强制终止时，临时文件可能留下。

函数会把序列化后的文本保存在内存中，适合小型状态文档。处理大文件时，应继续使用大型 CSV 流式读取方法。

4. 在自己的脚本中使用

从下载的模块导入函数：

from pathlib import Path
from atomic_json import atomic_write_json

Path("run").mkdir(exist_ok=True)
atomic_write_json("run/progress.json", {
    "schema_version": 1,
    "completed_batches": 12,
    "input_manifest": "inputs-v1.json",
})

对应的结果成功提交后，再保存进度。如果程序在写完结果、更新进度之前崩溃，下次运行可能重复处理一个批次。需要让重复操作不会造成错误，或使用能把结果与进度一起提交的存储方案。单个 JSON 文件的原子替换无法协调两者。

5. 检查失败后是否保留旧文件

演示程序检查了序列化失败。下面的独立检查则模拟临时文件已经写好、但替换操作失败的情况：

from pathlib import Path
from tempfile import TemporaryDirectory
from unittest.mock import patch
from atomic_json import atomic_write_json

with TemporaryDirectory() as directory:
    target = Path(directory) / "progress.json"
    atomic_write_json(target, {"completed_batches": 1})
    before = target.read_bytes()
    with patch("atomic_json.os.replace", side_effect=OSError("simulated failure")):
        try:
            atomic_write_json(target, {"completed_batches": 2})
        except OSError:
            pass
        else:
            raise AssertionError("Expected replacement to fail")
    assert target.read_bytes() == before
    assert list(Path(directory).iterdir()) == [target]
    print("PASS: old file preserved; temporary file removed")

这是注入错误的测试，不是断电测试。先调用 flush，再调用 os.fsync，可将缓冲的文件数据交给操作系统的同步调用。这个函数没有在替换后同步父目录，不承诺完整的断电持久性。Linux 的 fsync 手册解释了为什么目录项需要单独考虑。

6. 排错与限制

现象或需求	处理方法
`FileNotFoundError`	保存前先创建父目录。
`PermissionError`	检查目录写权限，以及是否有其他程序阻止替换。
NaN 引发 `ValueError`	修正计算，或明确把缺失值编码为允许的 `None` 等值。
遗留临时文件	停止写入进程，检查它的 `.progress.json.*.tmp` 文件，只删除确认已无用的文件。
两个工作进程更新同一状态	使用明确的锁或数据库事务；仅替换文件仍会丢失更新。
其他用户需要读取文件	制定权限策略；此函数不会复制旧文件的权限或元数据。

POSIX 系统上的临时文件使用较严格的权限，替换后保留的是临时文件的权限。如果共享配置依赖原有的所有者、ACL 或权限模式，不应直接套用这个函数。它也不保存历史版本；需要回滚时，应另存带版本的检查点。

7. 小结

先序列化小型 JSON 文档，在目标文件旁边写完新文件，再替换目标。正常保存和失败路径都要检查。把这个函数接入更大的处理流程时，还需要明确并发、恢复和持久性的要求。