ファイルへの書き込み中にプログラムがクラッシュすると、ファイルが中途半端な状態で残ることがある。アトミック（不可分）な書き込みを行えば、書き込みが完全に成功するか、まったく行われないかのどちらかを保証できる。
問題のあるコード
# ❌ クラッシュすると壊れたファイルが残る
with open('config.json', 'w') as f:
f.write('{"key": ')
# ここでクラッシュすると不完全な JSON が残る
f.write('"value"}')
この方法では、書き込み途中の不完全なファイルが残るリスクがある。
一時ファイルを使ったアトミック書き込み
基本的なパターンは、一時ファイルに書き込んでから、元のファイルにリネームすることだ。
import os
import tempfile
def atomic_write(filepath, content):
# 同じディレクトリに一時ファイルを作成
dir_name = os.path.dirname(filepath) or '.'
with tempfile.NamedTemporaryFile(
mode='w',
dir=dir_name,
delete=False
) as tmp:
tmp.write(content)
tmp_path = tmp.name
# リネーム（アトミックな操作）
os.replace(tmp_path, filepath)
# 使用例
atomic_write('config.json', '{"key": "value"}')
os.replace() は POSIX システムではアトミックな操作であり、書き込みが完全に成功するか、元のファイルがそのまま残るかのどちらかになる。
なぜリネームがアトミックなのか
ファイルシステムのリネーム操作は、メタデータ（ディレクトリエントリ）の更新だけで済むため、アトミックに実行できる。一方、ファイルの内容を直接書き換える場合は、複数のディスクブロックを更新する必要があり、途中で中断されるリスクがある。
同じファイルシステムに書く
os.replace() がアトミックに動作するには、一時ファイルと最終ファイルが同じファイルシステム上にある必要がある。異なるファイルシステム間では、コピー＋削除になり、アトミック性が失われる。
# ✅ 同じディレクトリに一時ファイルを作成
with tempfile.NamedTemporaryFile(dir='/same/directory', delete=False) as tmp:
pass
# ❌ /tmp は別のファイルシステムかもしれない
with tempfile.NamedTemporaryFile(delete=False) as tmp: # /tmp に作成される
pass
os.replace(tmp.name, '/home/user/data.txt') # 別ファイルシステムだと非アトミック
fsync でディスクに確実に書き込む
OS のバッファに書き込んだだけでは、電源断でデータが失われる可能性がある。fsync() を呼ぶと、データが物理ディスクに書き込まれる。
import os
import tempfile
def atomic_write_safe(filepath, content):
dir_name = os.path.dirname(filepath) or '.'
with tempfile.NamedTemporaryFile(
mode='w',
dir=dir_name,
delete=False
) as tmp:
tmp.write(content)
tmp.flush()
os.fsync(tmp.fileno()) # ディスクに確実に書き込む
tmp_path = tmp.name
os.replace(tmp_path, filepath)
# ディレクトリのメタデータも同期（より安全）
dir_fd = os.open(dir_name, os.O_RDONLY)
try:
os.fsync(dir_fd)
finally:
os.close(dir_fd)
atomicwrites ライブラリを使う
atomicwrites ライブラリを使えば、これらの処理を簡潔に書ける。
pip install atomicwrites
from atomicwrites import atomic_write
with atomic_write('config.json', overwrite=True) as f:
f.write('{"key": "value"}')
# 書き込みが成功した場合のみ、元のファイルが置き換わる
バックアップを残す
より安全にするため、上書き前にバックアップを残す方法もある。
import os
import shutil
def atomic_write_with_backup(filepath, content):
backup_path = filepath + '.bak'
# 既存ファイルがあればバックアップ
if os.path.exists(filepath):
shutil.copy2(filepath, backup_path)
# アトミック書き込み
atomic_write(filepath, content)
