64 lines
2.1 KiB
Python
64 lines
2.1 KiB
Python
import json
|
|
import os
|
|
import tempfile
|
|
import time
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
@dataclass
|
|
class CacheEntry:
|
|
value: Any
|
|
stale: bool
|
|
|
|
|
|
class FileCache:
|
|
def __init__(self, cache_dir: str, default_ttl_seconds: int) -> None:
|
|
self.cache_dir = Path(cache_dir)
|
|
self.cache_dir.mkdir(parents=True, exist_ok=True)
|
|
self.default_ttl_seconds = default_ttl_seconds
|
|
|
|
def _path_for(self, key: str) -> Path:
|
|
safe_key = "".join(c if c.isalnum() or c in ("-", "_") else "_" for c in key)
|
|
return self.cache_dir / f"{safe_key}.json"
|
|
|
|
def _read_raw(self, key: str) -> dict[str, Any] | None:
|
|
path = self._path_for(key)
|
|
if not path.exists():
|
|
return None
|
|
try:
|
|
return json.loads(path.read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError):
|
|
return None
|
|
|
|
def get_json(self, key: str, allow_stale: bool = False) -> CacheEntry | None:
|
|
payload = self._read_raw(key)
|
|
if payload is None:
|
|
return None
|
|
|
|
expires_at = float(payload.get("expires_at", 0))
|
|
value = payload.get("value")
|
|
stale = time.time() > expires_at
|
|
if stale and not allow_stale:
|
|
return None
|
|
return CacheEntry(value=value, stale=stale)
|
|
|
|
def set_json(self, key: str, value: Any, ttl_seconds: int | None = None) -> None:
|
|
ttl = ttl_seconds if ttl_seconds is not None else self.default_ttl_seconds
|
|
payload = {
|
|
"expires_at": time.time() + max(ttl, 0),
|
|
"value": value,
|
|
}
|
|
self._atomic_write_json(self._path_for(key), payload)
|
|
|
|
def _atomic_write_json(self, path: Path, payload: dict[str, Any]) -> None:
|
|
fd, tmp_name = tempfile.mkstemp(prefix="cache_", suffix=".tmp", dir=str(self.cache_dir))
|
|
try:
|
|
with os.fdopen(fd, "w", encoding="utf-8") as tmp_file:
|
|
json.dump(payload, tmp_file, separators=(",", ":"))
|
|
os.replace(tmp_name, path)
|
|
finally:
|
|
if os.path.exists(tmp_name):
|
|
os.remove(tmp_name)
|