61 lines
1.7 KiB
Python
61 lines
1.7 KiB
Python
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
import os
|
|
import time
|
|
|
|
# TTL for cache entries in seconds (24 hours)
|
|
CACHE_TTL = 24 * 60 * 60
|
|
|
|
|
|
CACHE_DIR = Path(__file__).resolve().parents[1] / 'cache'
|
|
CACHE_DIR.mkdir(parents=True, exist_ok=True)
|
|
|
|
|
|
def _key_to_filename(key: str) -> Path:
|
|
h = hashlib.sha256(key.encode('utf-8')).hexdigest()
|
|
return CACHE_DIR / f'{h}.json'
|
|
|
|
|
|
def read_cache(key: str) -> Any:
|
|
# avoid returning cached values during pytest runs to keep tests deterministic
|
|
if os.environ.get('PYTEST_CURRENT_TEST'):
|
|
return None
|
|
path = _key_to_filename(key)
|
|
if not path.exists():
|
|
return None
|
|
try:
|
|
with path.open('r', encoding='utf-8') as f:
|
|
payload = json.load(f)
|
|
# payload expected to be {'created_at': <ts>, 'data': <actual>}
|
|
created = payload.get('created_at')
|
|
if created is None:
|
|
return payload.get('data', None)
|
|
# expire after TTL
|
|
if (time.time() - created) > CACHE_TTL:
|
|
try:
|
|
path.unlink()
|
|
except Exception:
|
|
pass
|
|
return None
|
|
return payload.get('data', None)
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def write_cache(key: str, data: Any) -> None:
|
|
# avoid writing cache during pytest runs to prevent test cross-talk
|
|
if os.environ.get('PYTEST_CURRENT_TEST'):
|
|
return
|
|
path = _key_to_filename(key)
|
|
tmp = path.with_suffix('.tmp')
|
|
try:
|
|
payload = {'created_at': time.time(), 'data': data}
|
|
with tmp.open('w', encoding='utf-8') as f:
|
|
json.dump(payload, f)
|
|
tmp.replace(path)
|
|
except Exception:
|
|
if tmp.exists():
|
|
tmp.unlink()
|