Python 标准库原子替换 JSON 配置：让失败更新不覆盖旧文件

配置文件经常被当作“读出来、改字典、直接写回去”的小问题。真正麻烦的是写回过程不是一个瞬间：程序可能在序列化、写入或替换之前失败。若直接以 w 模式打开目标文件，旧文件可能已经被截断；下一次启动看到的就不是上一份有效配置，而是半份 JSON 或空文件。

本文实现一个只依赖 Python 标准库的 atomic_write_json。它先在目标文件所在目录创建临时文件，完成 JSON 序列化、刷新和同步，再用 os.replace 一次替换目标路径。本文所有输出和结论都来自本地固定演示与验证脚本，不涉及云端、网络、性能压测或跨平台故障承诺。

一、先定义“失败时不动旧配置”

本例把一次更新分成三个可观察阶段：

1. 在目标目录创建临时文件。
2. 把新对象完整编码到临时文件，并完成 flush、os.fsync。
3. 调用 os.replace 让临时文件成为目标文件。

只有第三步成功，目标路径才应该看到新版本。如果新对象含有不能被 JSON 编码的值，例如 object()，错误应发生在临时文件上，旧配置仍可读取。

这不是把任何写入都变成事务，也不是数据库替代品。它只把“目标路径上的文件内容切换”安排到最后，并清理失败留下的临时文件。进程在写临时文件期间崩溃时，旧目标文件不会被这段代码主动截断；临时文件是否由操作系统保留下来，取决于崩溃时机和文件系统，正常异常路径则由 finally 清理。

二、为什么临时文件必须和目标文件同目录

os.replace(source, target) 在同一文件系统内替换最直接。把临时文件放到系统默认临时目录，再替换到配置目录，可能跨越文件系统边界，行为依赖环境，甚至直接失败。NamedTemporaryFile 传入 dir=path.parent，让临时文件与目标文件使用同一个目录。

临时文件名带有目标文件名的前缀和 .tmp 后缀，便于本地排查；它不是文章代码对外暴露的协议，也不能代替权限控制。delete=False 是为了在关闭文件后由 os.replace 接管路径。Windows 等环境下，关闭临时文件再替换尤其重要。

三、完整演示代码

把以下代码保存为 articles/atomic-json-config-demo.py。它只使用 Python 标准库，Python 3.10 或更高版本即可运行。无参数执行时，脚本在临时目录中先写入旧配置，故意尝试序列化不可编码对象，再执行一次成功更新。

"""Atomically replace a JSON configuration file with standard-library APIs."""
from __future__ import annotations

import json
import tempfile
from pathlib import Path
from typing import Any
import os


def atomic_write_json(path: Path, value: Any) -> None:
    """Write JSON beside path, then replace path only after the write succeeds."""
    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)
    temporary: Path | None = None
    try:
        with tempfile.NamedTemporaryFile(
            mode="w",
            encoding="utf-8",
            dir=path.parent,
            prefix=f".{path.name}.",
            suffix=".tmp",
            delete=False,
        ) as stream:
            temporary = Path(stream.name)
            json.dump(value, stream, ensure_ascii=False, indent=2, sort_keys=True)
            stream.write("\n")
            stream.flush()
            os.fsync(stream.fileno())
        os.replace(temporary, path)
        temporary = None
    finally:
        if temporary is not None:
            temporary.unlink(missing_ok=True)


def read_json(path: Path) -> dict[str, Any]:
    return json.loads(path.read_text(encoding="utf-8"))


def temporary_count(directory: Path, name: str) -> int:
    return sum(1 for item in directory.glob(f".{name}.*.tmp") if item.is_file())


def main() -> None:
    with tempfile.TemporaryDirectory() as directory_name:
        directory = Path(directory_name)
        config = directory / "service.json"
        original = {"enabled": False, "retries": 2, "service": "api"}
        atomic_write_json(config, original)

        try:
            atomic_write_json(config, {"enabled": object()})
        except TypeError as error:
            failed_update = {
                "error_type": type(error).__name__,
                "old_config_preserved": read_json(config) == original,
                "temporary_files_remaining": temporary_count(directory, config.name),
            }
        else:
            raise AssertionError("the non-serializable value unexpectedly succeeded")

        updated = {"enabled": True, "retries": 4, "service": "api"}
        atomic_write_json(config, updated)
        successful_update = {
            "config": read_json(config),
            "temporary_files_remaining": temporary_count(directory, config.name),
        }
        print(json.dumps({
            "failed_update": failed_update,
            "successful_update": successful_update,
        }, ensure_ascii=False, indent=2, sort_keys=True))


if __name__ == "__main__":
    main()

这里有几个关键点。

• json.dump 仍然可能失败，所以必须在 os.replace 之前执行；不能先替换再验证。
• flush 把 Python 缓冲区内容交给操作系统，os.fsync 请求把文件内容同步到存储设备。它会增加写入成本，但比只调用 flush 有更明确的持久化意图。
• os.replace 成功后，把 temporary 设为 None，避免 finally 再删除已经变成目标路径的文件。
• finally 只清理仍然存在的临时路径。missing_ok=True 让清理动作可以重复执行。
• ensure_ascii=False 保留非 ASCII 文本，sort_keys=True 让演示中的序列化顺序稳定；这两个选项是本例输出约定，不是原子替换的必要条件。

四、运行结果：失败更新保留旧值

在项目根目录运行：

"C:/Users/33872/.dsh/dsh-runtimes/dsh-primary-runtime/dependencies/python/python.exe" -B articles/atomic-json-config-demo.py

精确标准输出如下，末尾有一个换行：

{
  "failed_update": {
    "error_type": "TypeError",
    "old_config_preserved": true,
    "temporary_files_remaining": 0
  },
  "successful_update": {
    "config": {
      "enabled": true,
      "retries": 4,
      "service": "api"
    },
    "temporary_files_remaining": 0
  }
}

失败分支传入了不可序列化的 object()。TypeError 出现时，目标文件仍然等于 original，并且正常异常路径没有留下匹配的 .tmp 文件。成功分支再写入新配置，读取结果变为 enabled: true、retries: 4，临时文件计数仍为 0。

这个实验验证的是文件替换顺序和异常清理，不是“任何崩溃都能恢复”的证明。它也没有测量 fsync 的耗时，更没有证明目录元数据已经在断电后持久化。若业务需要更强的断电语义，还要根据目标操作系统和文件系统验证目录同步、权限、备份策略和恢复流程；不能仅凭这段示例宣称完整的耐久事务。

五、把边界写进调用约定

原子替换只处理“新文件准备好后如何切换”。调用方仍然要明确以下边界：

尤其要注意并发。两个进程都先读取旧配置、各自修改后调用本函数，最后一个成功替换的版本仍可能覆盖前一个版本的业务修改。原子替换避免的是半份文件，不是 lost update；需要合并语义时，应在同一个锁或数据库事务中完成读取、检查和写入。

六、如何做本地验证

验证脚本会重复运行演示，比较两次 stdout 字节；导入函数后检查空值、非 ASCII 字符、失败清理和成功替换；还会读取两篇文章，确认都嵌入了与演示文件完全一致的代码和 stdout。期望结果写成验证器中的 Python 字面量，不从演示输出反推，因此演示即使被改成自洽的错误结果，也会被独立期望值拦住。

在项目根目录运行：

"C:/Users/33872/.dsh/dsh-runtimes/dsh-primary-runtime/dependencies/python/python.exe" -B articles/atomic-json-config-validate.py

验证器会生成：

• articles/atomic-json-config-stdout.txt：本次演示的原始 stdout。
• articles/atomic-json-config-validation.json：Python 版本、平台、检查项、哈希和验证范围。

验证范围明确限定为“本地 Python 执行”。它没有访问华为云、浏览器、网络或外部服务，也没有把一次本地成功运行扩展成线上可靠性承诺。

七、结语

对单个 JSON 配置文件而言，NamedTemporaryFile、json.dump、flush、os.fsync 和 os.replace 足以组成一条清晰的写入路径：先准备完整的新文件，再切换目标路径。这个顺序让序列化失败不会先破坏旧配置，并且可以用固定输出验证失败与成功两条路径。

但它的边界同样清楚：原子替换不是并发锁，不是数据库事务，也不是跨平台断电恢复的完整证明。把这些边界写进调用约定，再用本地验证器持续检查，才是这类小工具适合进入项目的前提。