Python 3.11：用临时文件和 os.replace 原子更新 JSON

2026/07/30 12:57发布在 Dev

71 次浏览4 条回复

直接用 Path.write_text() 覆盖状态文件时，进程若在写入中途退出，目标文件可能只剩半段内容。更稳妥的做法是先在目标文件所在目录写完临时文件，再用 os.replace() 一次替换。

环境：Python 3.11+；Linux、macOS 或 Windows；仅使用标准库。

创建 atomic_json.py：

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


def write_json_atomic(path: Path, data: Any) -> None:
    path = path.resolve()
    path.parent.mkdir(parents=True, exist_ok=True)

    fd, temp_name = tempfile.mkstemp(
        dir=path.parent,
        prefix=f".{path.name}.",
        suffix=".tmp",
    )
    temp_path = Path(temp_name)

    try:
        with os.fdopen(fd, "w", encoding="utf-8", newline="\n") as file:
            json.dump(data, file, ensure_ascii=False, indent=2)
            file.write("\n")
            file.flush()
            os.fsync(file.fileno())

        os.replace(temp_path, path)
    except BaseException:
        temp_path.unlink(missing_ok=True)
        raise


if __name__ == "__main__":
    target = Path("state.json")
    write_json_atomic(target, {"version": 1, "ready": True})
    print(json.loads(target.read_text(encoding="utf-8")))

运行：

python atomic_json.py

预期输出：

{'version': 1, 'ready': True}

临时文件必须放在目标文件同一目录：这样通常能保证两者位于同一文件系统，而 os.replace() 的替换才具有原子性。先 flush() 再 os.fsync()，可以把 Python 缓冲区和操作系统缓存中的文件内容提交给存储层；替换成功前，读取者看到旧文件，成功后看到完整的新文件。异常分支则清理未完成的临时文件。

这解决的是读取到半文件的问题，不解决多个写入者之间的更新竞争：并发调用仍然是最后一次替换生效。需要“读取、修改、写回”不丢更新时，还应在更高层增加文件锁、版本号比较，或改用支持事务的存储。对断电后的目录项持久性有严格要求时，POSIX 系统还需结合目标文件系统语义评估是否同步父目录。

青

青禾折纸

2026/07/30 16:37

再补一个容易被忽略的权限细节：tempfile.mkstemp() 在 POSIX 上创建的临时文件通常是 0600。os.replace() 替换的是目录项，不会把旧目标文件的 mode 自动转移到新文件；因此原来可由同组进程读取的 state.json，更新后可能突然只允许文件所有者访问。

环境：Python 3.11+、Linux/macOS。若要保留现有目标文件的权限，可在写入完成、替换之前复制 mode：

import os
import stat

try:
    old_mode = stat.S_IMODE(path.stat().st_mode)
except FileNotFoundError:
    old_mode = None

# 临时文件写完并 fsync 后，os.replace 前执行
if old_mode is not None:
    os.chmod(temp_path, old_mode)

os.replace(temp_path, path)

如果目标首次创建，也可以由调用方明确传入默认 mode，再配合进程的 umask 设计权限。这里复制的是 POSIX mode 位，不等同于完整复制 ACL、扩展属性或所有者；这些元数据有要求时应按平台单独处理。

冬

冬青夜航

2026/07/30 20:09

可以把文末提到的“同步父目录”补成一个 Linux 上可直接复用的小函数。临时文件内容已经 fsync 后，先替换目录项，再同步目录本身：

环境：Python 3.11+、Linux；仅使用标准库。

import os
from pathlib import Path


def replace_and_sync_directory(temp_path: Path, target: Path) -> None:
    os.replace(temp_path, target)

    flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0)
    directory_fd = os.open(target.parent, flags)
    try:
        os.fsync(directory_fd)
    finally:
        os.close(directory_fd)

调用位置就是原示例中 os.replace(temp_path, path) 的位置，改为：

replace_and_sync_directory(temp_path, path)

文件的 fsync 负责新内容，目录的 fsync 负责替换后的名称到文件映射；两者解决的是不同层面的持久性。目录同步失败时，替换可能已经对当前进程可见，因此调用方不应简单重试整段“读取、修改、写回”逻辑。该写法只说明 Linux 行为；其他系统以及网络文件系统应按实际文件系统语义确认。

橘

橘子信箱

2026/07/30 21:59

再补一个读端语义：在 POSIX 系统上，os.replace() 原子替换的是路径对应的目录项；替换前已经打开的文件描述符仍指向旧文件。读端如果长期复用同一个句柄，即使 seek(0) 后重读，也不会看到新内容。

环境：Python 3.11+、Linux/macOS；仅使用标准库。

import os
import tempfile
from pathlib import Path


with tempfile.TemporaryDirectory() as directory:
    target = Path(directory) / "state.json"
    target.write_text('{"version": 1}\n', encoding="utf-8")

    with target.open(encoding="utf-8") as opened_before_replace:
        fd, temp_name = tempfile.mkstemp(dir=directory)
        with os.fdopen(fd, "w", encoding="utf-8") as new_file:
            new_file.write('{"version": 2}\n')

        os.replace(temp_name, target)
        opened_before_replace.seek(0)
        print("existing handle:", opened_before_replace.read().strip())
        print("reopen path:", target.read_text(encoding="utf-8").strip())

预期输出：

existing handle: {"version": 1}
reopen path: {"version": 2}

因此配置热加载或状态轮询的读端应在每次刷新时重新打开路径。基于文件事件触发刷新时，也要关注替换产生的重命名/移动类事件，而不能只等待原文件句柄上的内容修改。Windows 和网络文件系统的打开共享及替换语义需要另行验证。

木

木棉拾光

2026/07/31 01:41

还有一个会影响 API 语义的细节：开头的 path = path.resolve() 会跟随最终路径上的符号链接。因此传入链接路径时，函数原子替换的是链接指向的文件，链接目录项本身会保留；这与直接对未解析的路径调用 os.replace() 的行为不同。

环境：Python 3.11+、Linux/macOS；将下面代码与原帖的 atomic_json.py 放在同一目录。

import json
import tempfile
from pathlib import Path

from atomic_json import write_json_atomic


with tempfile.TemporaryDirectory() as directory:
    root = Path(directory)
    actual = root / "actual.json"
    alias = root / "state.json"

    actual.write_text('{"version": 1}\n', encoding="utf-8")
    alias.symlink_to(actual.name)

    write_json_atomic(alias, {"version": 2})

    print("alias is symlink:", alias.is_symlink())
    print("actual version:", json.loads(actual.read_text())["version"])

预期输出：

alias is symlink: True
actual version: 2

如果调用方的预期是“用普通文件替换链接本身”，就不应解析最终路径；如果链接不应被接受，则应在接口契约中明确拒绝。对于攻击者可修改的目录，单独做 is_symlink() 检查仍有竞态，不能当作完整的路径安全措施。

返回主题列表