79 lines
2.5 KiB
Python
79 lines
2.5 KiB
Python
"""File-based cache para data fetchers. JSON on disk con TTL.
|
|
|
|
Estructura:
|
|
.cache/data_fetchers/<namespace>_<hash16>.json
|
|
|
|
Cada entry:
|
|
{"cached_at": <epoch_seconds>, "key": "<original_key>", "data": {...}}
|
|
|
|
TTL se evalua en get() — si la entrada esta vencida, devuelve None
|
|
(no la borra; la sobreescribe el siguiente set()).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
|
|
class FileCache:
|
|
def __init__(self, cache_dir: str | Path):
|
|
self.cache_dir = Path(cache_dir)
|
|
self.cache_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
def _path(self, namespace: str, key: str) -> Path:
|
|
safe_key = hashlib.sha1(key.encode("utf-8")).hexdigest()[:16]
|
|
safe_ns = "".join(c if c.isalnum() else "_" for c in namespace)[:24]
|
|
return self.cache_dir / f"{safe_ns}_{safe_key}.json"
|
|
|
|
def get(self, namespace: str, key: str, ttl_days: float) -> Optional[dict]:
|
|
"""Devuelve el dict cacheado si existe y no esta vencido. Sino None."""
|
|
p = self._path(namespace, key)
|
|
if not p.exists():
|
|
return None
|
|
try:
|
|
with p.open(encoding="utf-8") as f:
|
|
entry = json.load(f)
|
|
cached_at = entry.get("cached_at", 0)
|
|
age_days = (time.time() - cached_at) / 86400.0
|
|
if age_days > ttl_days:
|
|
return None
|
|
return entry.get("data")
|
|
except (json.JSONDecodeError, OSError):
|
|
return None
|
|
|
|
def set(self, namespace: str, key: str, data: dict) -> None:
|
|
"""Guarda data al cache. Errores de escritura son silenciados (non-fatal)."""
|
|
p = self._path(namespace, key)
|
|
entry = {
|
|
"cached_at": time.time(),
|
|
"namespace": namespace,
|
|
"key": key,
|
|
"data": data,
|
|
}
|
|
try:
|
|
p.write_text(
|
|
json.dumps(entry, ensure_ascii=False, indent=2),
|
|
encoding="utf-8",
|
|
)
|
|
except OSError:
|
|
pass # cache failures are non-fatal
|
|
|
|
def clear(self, namespace: Optional[str] = None) -> int:
|
|
"""Borra entradas de cache. Si namespace, solo de esa namespace.
|
|
|
|
Devuelve cantidad de archivos borrados.
|
|
"""
|
|
count = 0
|
|
pattern = f"{namespace}_*.json" if namespace else "*.json"
|
|
for p in self.cache_dir.glob(pattern):
|
|
try:
|
|
p.unlink()
|
|
count += 1
|
|
except OSError:
|
|
pass
|
|
return count
|