"""Atomic file write utilities.""" import json import logging import os import tempfile from pathlib import Path logger = logging.getLogger(__name__) def atomic_write_json(file_path: Path, data: dict, indent: int = 2) -> None: """Write JSON data to file atomically via temp file + rename. Prevents data corruption if the process crashes or loses power mid-write. The rename operation is atomic on most filesystems. """ file_path = Path(file_path) file_path.parent.mkdir(parents=True, exist_ok=True) # Write to a temp file in the same directory (same filesystem for atomic rename) fd, tmp_path = tempfile.mkstemp( dir=file_path.parent, prefix=f".{file_path.stem}_", suffix=".tmp", ) try: with os.fdopen(fd, "w", encoding="utf-8") as f: json.dump(data, f, indent=indent, ensure_ascii=False) os.replace(tmp_path, file_path) except BaseException: # Clean up temp file on any error try: os.unlink(tmp_path) except OSError as e: logger.debug("Failed to clean up temp file %s: %s", tmp_path, e) pass raise