107 lines
3.4 KiB
Python
107 lines
3.4 KiB
Python
"""Zero-dependency in-process TTL cache for hot endpoints.
|
|
|
|
Designed for read-heavy, mostly-idempotent endpoints like the admin report
|
|
pages (``/api/reports/stats-corporate``, ``/api/reports/student-performance``)
|
|
and the AI narrative generator. A single LRU-ish dict per worker with a hard
|
|
time bound is usually enough to absorb the dashboard refresh storm caused by
|
|
an admin tabbing between pages.
|
|
|
|
**Not** a substitute for Redis — cross-worker consistency and bounded memory
|
|
are out of scope. If ops ever needs those guarantees, swap the internal dict
|
|
for a ``redis.Redis.setex`` call; the public API (``memoize_ttl`` and
|
|
``invalidate``) stays identical.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import logging
|
|
import threading
|
|
import time
|
|
from functools import wraps
|
|
from typing import Any, Callable
|
|
|
|
_logger = logging.getLogger(__name__)
|
|
|
|
_cache: dict[str, tuple[float, Any]] = {}
|
|
_lock = threading.Lock()
|
|
_MAX_ENTRIES = 512
|
|
|
|
|
|
def _make_key(namespace: str, args: tuple, kwargs: dict) -> str:
|
|
payload = json.dumps([args, kwargs], sort_keys=True, default=str)
|
|
digest = hashlib.md5(payload.encode("utf-8")).hexdigest()
|
|
return f"{namespace}:{digest}"
|
|
|
|
|
|
def get(key: str):
|
|
entry = _cache.get(key)
|
|
if not entry:
|
|
return None
|
|
expiry, value = entry
|
|
if expiry < time.time():
|
|
with _lock:
|
|
_cache.pop(key, None)
|
|
return None
|
|
return value
|
|
|
|
|
|
def put(key: str, value, ttl_seconds: int) -> None:
|
|
with _lock:
|
|
if len(_cache) >= _MAX_ENTRIES:
|
|
oldest = sorted(_cache.items(), key=lambda kv: kv[1][0])[: _MAX_ENTRIES // 4]
|
|
for k, _v in oldest:
|
|
_cache.pop(k, None)
|
|
_cache[key] = (time.time() + max(1, ttl_seconds), value)
|
|
|
|
|
|
def invalidate(namespace: str | None = None) -> int:
|
|
"""Drop every cached entry, or just entries under ``namespace``.
|
|
|
|
Returns the number of removed entries. Callers should invoke this from
|
|
write endpoints that affect the cached read so subsequent reads observe
|
|
the new state.
|
|
"""
|
|
removed = 0
|
|
with _lock:
|
|
if namespace is None:
|
|
removed = len(_cache)
|
|
_cache.clear()
|
|
return removed
|
|
prefix = f"{namespace}:"
|
|
keys = [k for k in _cache if k.startswith(prefix)]
|
|
for k in keys:
|
|
_cache.pop(k, None)
|
|
removed = len(keys)
|
|
return removed
|
|
|
|
|
|
def memoize_ttl(namespace: str, ttl_seconds: int = 30):
|
|
"""Decorator: cache ``func(*args, **kwargs)`` for ``ttl_seconds`` per key.
|
|
|
|
The key is derived from the namespace + a stable JSON dump of the args,
|
|
so callers don't need to worry about mutable keyword order or unhashable
|
|
defaults. JWT decorator should run *before* this one so unauthenticated
|
|
traffic never hits the cache.
|
|
"""
|
|
def decorator(func: Callable):
|
|
@wraps(func)
|
|
def wrapper(*args, **kwargs):
|
|
try:
|
|
key = _make_key(namespace, args, kwargs)
|
|
except Exception:
|
|
return func(*args, **kwargs)
|
|
cached = get(key)
|
|
if cached is not None:
|
|
return cached
|
|
value = func(*args, **kwargs)
|
|
try:
|
|
put(key, value, ttl_seconds)
|
|
except Exception:
|
|
_logger.debug("cache put failed for %s", namespace, exc_info=True)
|
|
return value
|
|
wrapper._encoach_cache_namespace = namespace
|
|
return wrapper
|
|
return decorator
|