import json
import os
import tempfile
from pathlib import Path
from typing import Any
def atomic_write_json(path: Path, data: Any, *, fsync: bool = True) -> None:
dir_path = path.parent
dir_path.mkdir(parents=True, exist_ok=True)
fd, tmp_path = tempfile.mkstemp(dir=str(dir_path), suffix=".tmp")
try:
with os.fdopen(fd, "w", encoding="utf-8") as f:
json.dump(data, f)
f.flush()
if fsync:
os.fsync(f.fileno())
os.replace(tmp_path, str(path))
except Exception:
try:
os.unlink(tmp_path)
except OSError:
pass
raise
def atomic_write_text(path: Path, content: str, *, fsync: bool = True) -> None:
dir_path = path.parent
dir_path.mkdir(parents=True, exist_ok=True)
fd, tmp_path = tempfile.mkstemp(dir=str(dir_path), suffix=".tmp")
try:
with os.fdopen(fd, "w", encoding="utf-8") as f:
f.write(content)
f.flush()
if fsync:
os.fsync(f.fileno())
os.replace(tmp_path, str(path))
except Exception:
try:
os.unlink(tmp_path)
except OSError:
pass
raise
def atomic_write_binary(path: Path, data: bytes, *, fsync: bool = True) -> None:
dir_path = path.parent
dir_path.mkdir(parents=True, exist_ok=True)
fd, tmp_path = tempfile.mkstemp(dir=str(dir_path), suffix=".tmp")
try:
with os.fdopen(fd, "wb") as f:
f.write(data)
f.flush()
if fsync:
os.fsync(f.fileno())
os.replace(tmp_path, str(path))
except Exception:
try:
os.unlink(tmp_path)
except OSError:
pass
raise
def fsync_path(path: Path) -> None:
with open(path, "rb") as f:
os.fsync(f.fileno())
