Atomically write text to a file using temp file + rename pattern. This ensures the file is never left in a corrupted/partial state if the process crashes mid-write. The rename operation is atomic on POSIX systems and nearly atomic on Windows (uses ReplaceFile API via Path.replace()
(path: Path, content: str, encoding: str = "utf-8")
| 9 | |
| 10 | |
| 11 | def atomic_write_text(path: Path, content: str, encoding: str = "utf-8") -> None: |
| 12 | """ |
| 13 | Atomically write text to a file using temp file + rename pattern. |
| 14 | |
| 15 | This ensures the file is never left in a corrupted/partial state if the |
| 16 | process crashes mid-write. The rename operation is atomic on POSIX systems |
| 17 | and nearly atomic on Windows (uses ReplaceFile API via Path.replace()). |
| 18 | |
| 19 | Args: |
| 20 | path: Path to the file to write |
| 21 | content: Text content to write |
| 22 | encoding: Text encoding (default: utf-8) |
| 23 | |
| 24 | Raises: |
| 25 | OSError/IOError: If the write fails and cleanup cannot complete |
| 26 | """ |
| 27 | # Ensure parent directory exists |
| 28 | path.parent.mkdir(parents=True, exist_ok=True) |
| 29 | |
| 30 | # Write to temp file first, then atomic rename |
| 31 | temp_path = path.with_suffix(path.suffix + ".tmp") |
| 32 | try: |
| 33 | with open(temp_path, "w", encoding=encoding) as f: |
| 34 | f.write(content) |
| 35 | f.flush() |
| 36 | # Ensure data is written to disk before rename |
| 37 | os.fsync(f.fileno()) |
| 38 | # Atomic rename - either completes fully or not at all |
| 39 | temp_path.replace(path) |
| 40 | except (OSError, IOError): |
| 41 | # Clean up temp file on error |
| 42 | try: |
| 43 | temp_path.unlink() |
| 44 | except (OSError, IOError): |
| 45 | pass |
| 46 | raise |
| 47 | |
| 48 | |
| 49 | def write_if_different(path: Path, content: str, mode: Optional[int] = None) -> bool: |
no test coverage detected