chore: initial import
This commit is contained in:
@@ -0,0 +1,27 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
from __future__ import annotations
|
||||
import json, os, shutil, tempfile
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
|
||||
def utc_now() -> str:
|
||||
return datetime.now(timezone.utc).isoformat(timespec="seconds")
|
||||
|
||||
|
||||
def atomic_json_write(path: str | Path, payload: Any, backup_dir: str | Path | None = None) -> None:
|
||||
target = Path(path)
|
||||
target.parent.mkdir(parents=True, exist_ok=True)
|
||||
if backup_dir and target.exists():
|
||||
bdir = Path(backup_dir); bdir.mkdir(parents=True, exist_ok=True)
|
||||
stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S%fZ")
|
||||
shutil.copy2(target, bdir / f"{target.stem}_{stamp}{target.suffix}")
|
||||
fd, tmp = tempfile.mkstemp(prefix=f".{target.name}.", suffix=".tmp", dir=target.parent)
|
||||
try:
|
||||
with os.fdopen(fd, "w", encoding="utf-8") as handle:
|
||||
json.dump(payload, handle, ensure_ascii=False, indent=2)
|
||||
handle.flush(); os.fsync(handle.fileno())
|
||||
os.replace(tmp, target)
|
||||
finally:
|
||||
if os.path.exists(tmp): os.unlink(tmp)
|
||||
Reference in New Issue
Block a user