Spaces:
Paused
Paused
File size: 1,246 Bytes
e1c927b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 | """File persistence under runs/rotr/ (platform convention: request-scoped
FastAPI over JSON files; no services). `set_rotr_root` is the test hook,
mirroring set_safety_root / set_nextgen_root etc."""
from __future__ import annotations
import json
import os
import threading
from typing import Any, Dict, List, Optional
_ROOT = os.path.join("runs", "rotr")
_ROOT_LOCK = threading.Lock()
def rotr_root() -> str:
return _ROOT
def set_rotr_root(path: str) -> None:
global _ROOT
with _ROOT_LOCK:
_ROOT = str(path)
def _path(*parts: str) -> str:
return os.path.join(rotr_root(), *parts)
def write_json(obj: Any, *parts: str) -> str:
path = _path(*parts)
os.makedirs(os.path.dirname(path), exist_ok=True)
with open(path, "w") as f:
json.dump(obj, f, indent=1, default=str)
return path
def read_json(*parts: str) -> Optional[Dict]:
path = _path(*parts)
if not os.path.exists(path):
return None
with open(path) as f:
return json.load(f)
def list_dir(*parts: str) -> List[str]:
path = _path(*parts)
if not os.path.isdir(path):
return []
return sorted(os.listdir(path))
def exists(*parts: str) -> bool:
return os.path.exists(_path(*parts))
|