给 Python 配置文件做原子替换:写到一半失败时,旧文件还能用吗?
【摘要】 用 open(path, "w") 保存配置时,文件一打开就会被截断。如果序列化失败,或者另一个进程恰好在写入途中读取,它可能只拿到一段不能解析的 JSON。对一个由单个进程更新、多个进程读取的本地配置,更实用的发布点是一次同目录的 os.replace:先把新内容写完整,再替换目标路径。 把写入与发布分开下面的函数接受一个已有父目录下的文件路径。它先序列化,避免不支持的对象触及磁盘;临时文...
用 open(path, "w") 保存配置时,文件一打开就会被截断。如果序列化失败,或者另一个进程恰好在写入途中读取,它可能只拿到一段不能解析的 JSON。对一个由单个进程更新、多个进程读取的本地配置,更实用的发布点是一次同目录的 os.replace:先把新内容写完整,再替换目标路径。
把写入与发布分开
下面的函数接受一个已有父目录下的文件路径。它先序列化,避免不支持的对象触及磁盘;临时文件和目标位于同一目录,避免跨文件系统替换。先 flush 把 Python 缓冲区的数据交给系统,再 fsync 临时文件;两个步骤都成功后才关闭临时文件并替换目标。这段示例只讨论本地文件系统,不需要云资源或第三方库。
import json
import os
from pathlib import Path
from tempfile import NamedTemporaryFile
def write_config(path: Path, config: dict) -> None:
text = json.dumps(config, ensure_ascii=False, allow_nan=False, indent=2) + "\n"
temporary_path = None
try:
with NamedTemporaryFile(
mode="w", encoding="utf-8", newline="\n", dir=path.parent,
prefix=f".{path.name}.", suffix=".tmp", delete=False,
) as stream:
temporary_path = Path(stream.name)
stream.write(text)
stream.flush()
os.fsync(stream.fileno())
os.replace(temporary_path, path)
finally:
if temporary_path is not None:
temporary_path.unlink(missing_ok=True)
在支持同目录原子替换的文件系统上,从目标路径重新打开文件的读者看到的是旧文件或新文件,不会看到临时文件写到一半的内容。已经持有旧文件句柄的读者不会自动切换到新版本;需要刷新时应重新打开路径。若序列化、写入或替换在发布前失败,函数会报告异常,不应在调用方把这次保存标为成功。
用失败注入验证旧文件没有被截断
下面的测试可接在同一文件末尾运行。先验证 UTF-8 往返,再模拟替换失败,检查旧字节不变且临时文件已清理;最后验证正常更新和非法值被拒绝。模拟替换失败不等价于模拟整机断电。
from tempfile import TemporaryDirectory
from unittest.mock import patch
with TemporaryDirectory() as directory:
path = Path(directory) / "settings.json"
write_config(path, {"name": "相册", "version": 1})
original = path.read_bytes()
assert json.loads(original) == {"name": "相册", "version": 1}
assert "相册".encode("utf-8") in original
with patch.object(os, "replace", side_effect=PermissionError("模拟拒绝替换")):
try:
write_config(path, {"version": 2})
except PermissionError:
pass
else:
raise AssertionError("应传播保存失败")
assert path.read_bytes() == original
assert list(path.parent.iterdir()) == [path]
write_config(path, {"version": 2})
assert json.loads(path.read_bytes()) == {"version": 2}
try:
write_config(path, {"version": float("nan")})
except ValueError:
pass
else:
raise AssertionError("应拒绝非标准 JSON 数值")
assert json.loads(path.read_bytes()) == {"version": 2}
原子可见性不等于断电后一定持久
- 持久性:示例同步了新文件内容,没有同步父目录。在支持目录同步的 POSIX 文件系统上,如果还要求断电后尽可能保留新目录项,需要在替换后对父目录执行
fsync并处理失败;此时替换已经发生,目录同步失败不能被当作“旧版本仍在”。不要把本地实验的结果直接推广到网络文件系统。 - 并发写入:两名写入者都从旧配置读取并各自更新时,原子替换只保证读者不会看到半份文件,不能防止后写者覆盖先写者。若要保护“读取—修改—保存”全过程,应使用跨进程锁、版本检查或数据库事务。
- 权限与清理:替换的是新文件,不保证继承旧文件的权限、所有者或 ACL;如果配置要由其他账户读取,须明确设计权限。进程被强杀时
finally不会运行,清理残留临时文件也要避免误删另一个正在写入的进程的文件。
本文示例由 AI 辅助整理,使用本地 Python 标准库验证;没有操作真实业务配置。
【声明】本内容来自华为云开发者社区博主,不代表华为云及华为云开发者社区的观点和立场。转载时必须标注文章的来源(华为云社区)、文章链接、文章作者等基本信息,否则作者和本社区有权追究责任。如果您发现本社区中有涉嫌抄袭的内容,欢迎发送邮件进行举报,并提供相关证据,一经查实,本社区将立刻删除涉嫌侵权内容,举报邮箱:
cloudbbs@huaweicloud.com
- 点赞
- 收藏
- 关注作者
评论(0)