File size: 2,432 Bytes
af4851b | 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 56 57 58 59 60 61 62 63 64 65 66 67 68 | import shutil
import time
import json
from pathlib import Path
from typing import Any, Dict, List, Optional
from backend.types import SessionRecord
class SessionStore:
def __init__(self, root: Path, ttl_seconds: int) -> None:
self.root = Path(root)
self.ttl_seconds = ttl_seconds
self.root.mkdir(parents=True, exist_ok=True)
self._sessions: Dict[str, SessionRecord] = {}
def _now(self) -> float:
return time.time()
def ensure_session(self, session_id: str) -> SessionRecord:
now = self._now()
session_root = self.root / session_id
session_root.mkdir(parents=True, exist_ok=True)
for child in ("uploads", "previews", "renders", "exports"):
(session_root / child).mkdir(exist_ok=True)
existing = self._sessions.get(session_id)
if existing is None:
existing = SessionRecord(
session_id=session_id,
root=session_root,
created_at=now,
touched_at=now,
)
self._sessions[session_id] = existing
else:
existing.touched_at = now
return existing
def touch(self, session_id: str) -> SessionRecord:
session = self.ensure_session(session_id)
session.touched_at = self._now()
return session
def save_json(self, session_id: str, name: str, payload: Dict[str, Any]) -> Path:
session = self.touch(session_id)
path = session.root / name
path.write_text(json.dumps(payload, ensure_ascii=True, indent=2), encoding="utf-8")
return path
def load_json(self, session_id: str, name: str) -> Dict[str, Any]:
session = self.touch(session_id)
path = session.root / name
if not path.exists():
raise FileNotFoundError(f"Session asset not found: {name}")
return json.loads(path.read_text(encoding="utf-8"))
def cleanup_expired(self, now: Optional[float] = None) -> List[str]:
current = self._now() if now is None else now
removed: List[str] = []
for session_id, record in list(self._sessions.items()):
if current - record.touched_at <= self.ttl_seconds:
continue
if record.root.exists():
shutil.rmtree(record.root)
removed.append(session_id)
del self._sessions[session_id]
return removed
|