atomic.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 1 | """Atomic per-file writes per §K4.8.""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | import os |
| 6 | from pathlib import Path |
| 7 | |
| 8 | |
| 9 | class WriteFailure(Exception): |
| 10 | """Raised when an atomic write fails.""" |
| 11 | |
| 12 | def __init__(self, path: Path, cause: OSError) -> None: |
| 13 | self.path = path |
| 14 | self.cause = cause |
| 15 | super().__init__(f"write failed for {path}: {cause}") |
| 16 | |
| 17 | |
| 18 | def atomic_write_bytes(path: Path, data: bytes) -> None: |
| 19 | """Write ``data`` atomically via temp file + ``os.replace``.""" |
| 20 | path.parent.mkdir(parents=True, exist_ok=True) |
| 21 | tmp_path = path.with_name(f".{path.name}.overseer.tmp") |
| 22 | try: |
| 23 | tmp_path.write_bytes(data) |
| 24 | os.replace(tmp_path, path) |
| 25 | except OSError as exc: |
| 26 | if tmp_path.exists(): |
| 27 | try: |
| 28 | tmp_path.unlink() |
| 29 | except OSError: |
| 30 | pass |
| 31 | raise WriteFailure(path, exc) from exc |
| 32 | |
| 33 | |
| 34 | def atomic_write_text(path: Path, text: str) -> None: |
| 35 | """Write UTF-8 text atomically.""" |
| 36 | atomic_write_bytes(path, text.encode("utf-8")) |