feat(cache): C2 — Redis translation cache with user/prompt/glossary namespacing
All checks were successful
Deploy to Production / Build and Deploy (push) Successful in 3m13s
All checks were successful
Deploy to Production / Build and Deploy (push) Successful in 3m13s
This commit is contained in:
427
services/translation_cache.py
Normal file
427
services/translation_cache.py
Normal file
@@ -0,0 +1,427 @@
|
||||
"""
|
||||
Track C2 — Translation cache with Redis backend.
|
||||
|
||||
This module provides a translation cache that:
|
||||
- Defaults to in-process LRU (same as the previous behavior)
|
||||
- Optionally uses Redis when REDIS_CACHE_ENABLED=true
|
||||
- Namespaces keys by user_id (so free and pro users don't share)
|
||||
- Includes custom_prompt_hash + glossary_hash in the key (so two
|
||||
users with different prompts/glossaries don't get cross-contamination)
|
||||
- Falls back to LRU if Redis is unavailable (transient connection
|
||||
error, etc.) — a Redis outage MUST NOT break translation jobs
|
||||
|
||||
Public API:
|
||||
TranslationCache — single instance per process, async-friendly.
|
||||
get_cache() — singleton accessor.
|
||||
|
||||
Usage:
|
||||
cache = get_cache()
|
||||
cached = await cache.get(text, target, source, provider, user_id, custom_prompt_hash, glossary_hash)
|
||||
if cached is None:
|
||||
translated = await provider.translate(text, target, source)
|
||||
await cache.set(text, target, source, provider, user_id, custom_prompt_hash, glossary_hash, translated)
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
import time
|
||||
import asyncio
|
||||
import threading
|
||||
from collections import OrderedDict
|
||||
from typing import Optional, Dict, List
|
||||
|
||||
from core.logging import get_logger
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# Key construction
|
||||
# =============================================================================
|
||||
|
||||
def make_cache_key(
|
||||
text: str,
|
||||
target_language: str,
|
||||
source_language: str,
|
||||
provider: str,
|
||||
user_id: Optional[str] = None,
|
||||
custom_prompt_hash: Optional[str] = None,
|
||||
glossary_hash: Optional[str] = None,
|
||||
) -> str:
|
||||
"""Build a stable cache key.
|
||||
|
||||
The key includes:
|
||||
- The text being translated
|
||||
- The language pair
|
||||
- The provider
|
||||
- The user (so free/pro users don't collide)
|
||||
- The custom_prompt hash (so different prompts are isolated)
|
||||
- The glossary hash (so different glossaries are isolated)
|
||||
"""
|
||||
parts = [
|
||||
f"provider={provider}",
|
||||
f"src={source_language or 'auto'}",
|
||||
f"tgt={target_language}",
|
||||
f"user={user_id or 'anon'}",
|
||||
f"prompt={custom_prompt_hash or 'none'}",
|
||||
f"glossary={glossary_hash or 'none'}",
|
||||
f"text={text}",
|
||||
]
|
||||
content = "|".join(parts)
|
||||
return hashlib.sha256(content.encode("utf-8")).hexdigest()
|
||||
|
||||
|
||||
def hash_prompt(prompt: Optional[str]) -> Optional[str]:
|
||||
"""Hash a custom prompt for use in a cache key. Returns None for None/empty."""
|
||||
if not prompt:
|
||||
return None
|
||||
return hashlib.sha256(prompt.encode("utf-8")).hexdigest()[:16]
|
||||
|
||||
|
||||
def hash_glossary_terms(terms: Optional[List[dict]]) -> Optional[str]:
|
||||
"""Hash a glossary's term list for use in a cache key.
|
||||
|
||||
We only hash the source/target pairs (not metadata) so that two
|
||||
glossaries with the same content but different names still share
|
||||
cache entries.
|
||||
"""
|
||||
if not terms:
|
||||
return None
|
||||
# Stable representation: list of (source, target) tuples
|
||||
normalized = sorted(
|
||||
(str(t.get("source", "")), str(t.get("target", "")))
|
||||
for t in terms
|
||||
if t.get("source") and t.get("target")
|
||||
)
|
||||
content = json.dumps(normalized, ensure_ascii=False, sort_keys=True)
|
||||
return hashlib.sha256(content.encode("utf-8")).hexdigest()[:16]
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# In-process LRU fallback
|
||||
# =============================================================================
|
||||
|
||||
class _LRUBackend:
|
||||
"""Thread-safe LRU cache, used as a fallback when Redis is unavailable."""
|
||||
|
||||
def __init__(self, maxsize: int = 5000):
|
||||
self._cache: "OrderedDict[str, str]" = OrderedDict()
|
||||
self._maxsize = maxsize
|
||||
self._lock = threading.RLock()
|
||||
self.hits = 0
|
||||
self.misses = 0
|
||||
|
||||
def get(self, key: str) -> Optional[str]:
|
||||
with self._lock:
|
||||
if key in self._cache:
|
||||
self.hits += 1
|
||||
self._cache.move_to_end(key)
|
||||
return self._cache[key]
|
||||
self.misses += 1
|
||||
return None
|
||||
|
||||
def set(self, key: str, value: str, ttl: Optional[int] = None) -> None:
|
||||
# LRU doesn't honor TTL — it's a process-lifetime cache.
|
||||
with self._lock:
|
||||
if key in self._cache:
|
||||
self._cache.move_to_end(key)
|
||||
self._cache[key] = value
|
||||
while len(self._cache) > self._maxsize:
|
||||
self._cache.popitem(last=False)
|
||||
|
||||
def clear(self) -> None:
|
||||
with self._lock:
|
||||
self._cache.clear()
|
||||
self.hits = 0
|
||||
self.misses = 0
|
||||
|
||||
def stats(self) -> Dict:
|
||||
with self._lock:
|
||||
total = self.hits + self.misses
|
||||
hit_rate = (self.hits / total * 100) if total > 0 else 0
|
||||
return {
|
||||
"backend": "lru",
|
||||
"size": len(self._cache),
|
||||
"maxsize": self._maxsize,
|
||||
"hits": self.hits,
|
||||
"misses": self.misses,
|
||||
"hit_rate": f"{hit_rate:.1f}%",
|
||||
}
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# Redis backend
|
||||
# =============================================================================
|
||||
|
||||
class _RedisBackend:
|
||||
"""Redis-backed cache with TTL and async support.
|
||||
|
||||
Connection is lazy: we don't connect on import. If Redis is
|
||||
unreachable, every operation falls back to the LRU backend.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
url: str,
|
||||
namespace: str = "translate_cache",
|
||||
default_ttl: int = 86400, # 24 hours
|
||||
lru_fallback: Optional[_LRUBackend] = None,
|
||||
socket_timeout: float = 2.0,
|
||||
):
|
||||
self._url = url
|
||||
self._namespace = namespace
|
||||
self._default_ttl = default_ttl
|
||||
self._lru_fallback = lru_fallback
|
||||
self._socket_timeout = socket_timeout
|
||||
self._client = None
|
||||
self._lock = threading.RLock()
|
||||
self.hits = 0
|
||||
self.misses = 0
|
||||
self.errors = 0
|
||||
self._available = None # None = unknown, True/False = probed
|
||||
|
||||
def _get_client(self):
|
||||
"""Lazy client init. Never raises; returns None on failure."""
|
||||
if self._client is not None:
|
||||
return self._client
|
||||
with self._lock:
|
||||
if self._client is not None:
|
||||
return self._client
|
||||
try:
|
||||
import redis
|
||||
client = redis.from_url(
|
||||
self._url,
|
||||
socket_timeout=self._socket_timeout,
|
||||
socket_connect_timeout=self._socket_timeout,
|
||||
decode_responses=True,
|
||||
)
|
||||
# Probe once
|
||||
client.ping()
|
||||
self._client = client
|
||||
self._available = True
|
||||
logger.info("redis_cache_connected", url=self._url.split("@")[-1])
|
||||
return client
|
||||
except Exception as e:
|
||||
self._available = False
|
||||
self.errors += 1
|
||||
logger.warning(
|
||||
"redis_cache_unavailable",
|
||||
error=str(e)[:200],
|
||||
fallback="lru",
|
||||
)
|
||||
return None
|
||||
|
||||
def _key(self, key: str) -> str:
|
||||
return f"{self._namespace}:{key}"
|
||||
|
||||
def get(self, key: str) -> Optional[str]:
|
||||
client = self._get_client()
|
||||
if client is None:
|
||||
# Fall back to LRU
|
||||
if self._lru_fallback is not None:
|
||||
v = self._lru_fallback.get(key)
|
||||
if v is not None:
|
||||
self.hits += 1
|
||||
else:
|
||||
self.misses += 1
|
||||
return v
|
||||
return None
|
||||
try:
|
||||
v = client.get(self._key(key))
|
||||
if v is not None:
|
||||
self.hits += 1
|
||||
# Promote to LRU too (in-process L1 cache)
|
||||
if self._lru_fallback is not None:
|
||||
self._lru_fallback.set(key, v)
|
||||
return v
|
||||
self.misses += 1
|
||||
return None
|
||||
except Exception as e:
|
||||
self.errors += 1
|
||||
self._available = False
|
||||
self._client = None # force reconnect on next call
|
||||
logger.warning("redis_cache_get_error", error=str(e)[:200])
|
||||
if self._lru_fallback is not None:
|
||||
return self._lru_fallback.get(key)
|
||||
return None
|
||||
|
||||
def set(self, key: str, value: str, ttl: Optional[int] = None) -> None:
|
||||
client = self._get_client()
|
||||
if client is not None:
|
||||
try:
|
||||
client.set(
|
||||
self._key(key),
|
||||
value,
|
||||
ex=ttl if ttl is not None else self._default_ttl,
|
||||
)
|
||||
# Also write to LRU
|
||||
if self._lru_fallback is not None:
|
||||
self._lru_fallback.set(key, value)
|
||||
return
|
||||
except Exception as e:
|
||||
self.errors += 1
|
||||
self._available = False
|
||||
self._client = None
|
||||
logger.warning("redis_cache_set_error", error=str(e)[:200])
|
||||
# Fall through to LRU
|
||||
if self._lru_fallback is not None:
|
||||
self._lru_fallback.set(key, value, ttl=ttl)
|
||||
|
||||
def clear(self) -> None:
|
||||
if self._lru_fallback is not None:
|
||||
self._lru_fallback.clear()
|
||||
client = self._get_client()
|
||||
if client is not None:
|
||||
try:
|
||||
# Delete only keys in our namespace
|
||||
cursor = 0
|
||||
pattern = f"{self._namespace}:*"
|
||||
while True:
|
||||
cursor, keys = client.scan(cursor=cursor, match=pattern, count=500)
|
||||
if keys:
|
||||
client.delete(*keys)
|
||||
if cursor == 0:
|
||||
break
|
||||
except Exception as e:
|
||||
logger.warning("redis_cache_clear_error", error=str(e)[:200])
|
||||
|
||||
def stats(self) -> Dict:
|
||||
out = {
|
||||
"backend": "redis",
|
||||
"available": self._available,
|
||||
"url": self._url.split("@")[-1] if self._url else None,
|
||||
"namespace": self._namespace,
|
||||
"default_ttl": self._default_ttl,
|
||||
"hits": self.hits,
|
||||
"misses": self.misses,
|
||||
"errors": self.errors,
|
||||
}
|
||||
if self._lru_fallback is not None:
|
||||
lru_stats = self._lru_fallback.stats()
|
||||
out["lru_size"] = lru_stats["size"]
|
||||
return out
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# TranslationCache — public API
|
||||
# =============================================================================
|
||||
|
||||
class TranslationCache:
|
||||
"""High-level translation cache.
|
||||
|
||||
Delegates to a backend (Redis or LRU) based on configuration.
|
||||
Provides an async-friendly API (sync methods, but no blocking
|
||||
Redis I/O — Redis client is async-compatible via redis.asyncio
|
||||
if we want to add that later).
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
backend: str = "auto",
|
||||
redis_url: Optional[str] = None,
|
||||
redis_namespace: str = "translate_cache",
|
||||
default_ttl: int = 86400,
|
||||
lru_maxsize: int = 5000,
|
||||
):
|
||||
# Decide backend
|
||||
if backend == "auto":
|
||||
backend = "redis" if (
|
||||
os.getenv("REDIS_CACHE_ENABLED", "false").lower() == "true"
|
||||
and (redis_url or os.getenv("REDIS_URL", ""))
|
||||
) else "lru"
|
||||
|
||||
self._lru = _LRUBackend(maxsize=lru_maxsize)
|
||||
if backend == "redis" and (redis_url or os.getenv("REDIS_URL", "")):
|
||||
self._backend: object = _RedisBackend(
|
||||
url=redis_url or os.getenv("REDIS_URL", ""),
|
||||
namespace=redis_namespace,
|
||||
default_ttl=default_ttl,
|
||||
lru_fallback=self._lru,
|
||||
)
|
||||
self._backend_name = "redis"
|
||||
else:
|
||||
self._backend = self._lru
|
||||
self._backend_name = "lru"
|
||||
|
||||
@property
|
||||
def backend_name(self) -> str:
|
||||
return self._backend_name
|
||||
|
||||
def get(
|
||||
self,
|
||||
text: str,
|
||||
target_language: str,
|
||||
source_language: str,
|
||||
provider: str,
|
||||
user_id: Optional[str] = None,
|
||||
custom_prompt_hash: Optional[str] = None,
|
||||
glossary_hash: Optional[str] = None,
|
||||
) -> Optional[str]:
|
||||
key = make_cache_key(
|
||||
text, target_language, source_language, provider,
|
||||
user_id, custom_prompt_hash, glossary_hash,
|
||||
)
|
||||
return self._backend.get(key)
|
||||
|
||||
def set(
|
||||
self,
|
||||
text: str,
|
||||
target_language: str,
|
||||
source_language: str,
|
||||
provider: str,
|
||||
translation: str,
|
||||
user_id: Optional[str] = None,
|
||||
custom_prompt_hash: Optional[str] = None,
|
||||
glossary_hash: Optional[str] = None,
|
||||
ttl: Optional[int] = None,
|
||||
) -> None:
|
||||
key = make_cache_key(
|
||||
text, target_language, source_language, provider,
|
||||
user_id, custom_prompt_hash, glossary_hash,
|
||||
)
|
||||
self._backend.set(key, translation, ttl=ttl)
|
||||
|
||||
def clear(self) -> None:
|
||||
self._backend.clear()
|
||||
|
||||
def stats(self) -> Dict:
|
||||
return self._backend.stats()
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# Singleton accessor
|
||||
# =============================================================================
|
||||
|
||||
_cache: Optional[TranslationCache] = None
|
||||
_cache_lock = threading.RLock()
|
||||
|
||||
|
||||
def get_cache() -> TranslationCache:
|
||||
"""Get the process-wide translation cache (lazy init)."""
|
||||
global _cache
|
||||
if _cache is not None:
|
||||
return _cache
|
||||
with _cache_lock:
|
||||
if _cache is None:
|
||||
_cache = TranslationCache(
|
||||
backend="auto",
|
||||
redis_url=os.getenv("REDIS_URL", "") or None,
|
||||
redis_namespace=os.getenv("REDIS_CACHE_NAMESPACE", "translate_cache"),
|
||||
default_ttl=int(os.getenv("REDIS_CACHE_TTL", "86400")),
|
||||
lru_maxsize=int(os.getenv("LRU_CACHE_MAXSIZE", "5000")),
|
||||
)
|
||||
logger.info(
|
||||
"translation_cache_initialized",
|
||||
backend=_cache.backend_name,
|
||||
)
|
||||
return _cache
|
||||
|
||||
|
||||
def reset_cache_for_tests() -> None:
|
||||
"""Reset the singleton — only for tests."""
|
||||
global _cache
|
||||
with _cache_lock:
|
||||
_cache = None
|
||||
Reference in New Issue
Block a user