| """Publish complete files without replacing a usable file after a failed write.""" | |
| from contextlib import contextmanager | |
| import os | |
| from pathlib import Path | |
| import tempfile | |
| def atomic_output(path, mode="wb"): | |
| path = Path(path) | |
| path.parent.mkdir(parents=True, exist_ok=True) | |
| handle = tempfile.NamedTemporaryFile(mode=mode, dir=path.parent, | |
| prefix=path.name + ".", suffix=".partial", delete=False, | |
| **({"encoding": "utf-8"} if "b" not in mode else {})) | |
| temporary = Path(handle.name) | |
| try: | |
| with handle: | |
| yield handle | |
| handle.flush() | |
| os.fsync(handle.fileno()) | |
| os.replace(temporary, path) | |
| if os.name != "nt": | |
| descriptor = os.open(path.parent, os.O_RDONLY) | |
| try: | |
| os.fsync(descriptor) | |
| finally: | |
| os.close(descriptor) | |
| finally: | |
| temporary.unlink(missing_ok=True) | |