中文
devexamples

Pathlib Atomic Write Temp Rename

Write files atomically with pathlib by staging content to a temp file in the same directory and renaming over the target on success.

PythonData Integrity

Code

snippet
import os
import tempfile
from pathlib import Path


def atomic_write(path: Path, data: bytes) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)

    # 临时文件必须与目标同目录，rename 才能保证原子性
    fd, tmp_name = tempfile.mkstemp(
        dir=path.parent, prefix=f'.{path.name}.', suffix='.tmp'
    )
    tmp_path = Path(tmp_name)

    try:
        with os.fdopen(fd, 'wb') as fh:
            fh.write(data)
            fh.flush()
            os.fsync(fh.fileno())  # 确保数据先于元数据落盘
        os.replace(tmp_path, path)  # 原子替换，读者永远看不到半成品
    except BaseException:
        tmp_path.unlink(missing_ok=True)
        raise


def read_or_default(path: Path, default: bytes) -> bytes:
    try:
        return path.read_bytes()
    except FileNotFoundError:
        atomic_write(path, default)
        return default


config = Path('state/registry.json')
payload = b'{"version": 2, "entries": 42}'

atomic_write(config, payload)
print('size on disk:', read_or_default(config, b'{}'))

partial = Path('state/huge.bin')
atomic_write(partial, b'x' * 1024 * 1024)
print('big write done:', partial.stat().st_size)

Related Examples