84 lines
2.4 KiB
Python
84 lines
2.4 KiB
Python
"""
|
|
Módulo de caché Redis - Bots de Traducción
|
|
Proporciona una interfaz unificada de caché con Redis como backend
|
|
y fallback a memoria RAM si Redis no está disponible.
|
|
"""
|
|
import os
|
|
import redis
|
|
from utils.logger import discord_logger as log
|
|
|
|
_redis_client = None
|
|
|
|
def get_redis() -> redis.Redis | None:
|
|
"""Retorna cliente Redis, intentando conectar si no existe. Devuelve None si no disponible."""
|
|
global _redis_client
|
|
if _redis_client is not None:
|
|
return _redis_client
|
|
|
|
host = os.getenv("REDIS_HOST", "localhost")
|
|
port = int(os.getenv("REDIS_PORT", "6379"))
|
|
password = os.getenv("REDIS_PASSWORD", "translation_redis_secret")
|
|
db = int(os.getenv("REDIS_DB", "0"))
|
|
|
|
try:
|
|
client = redis.Redis(
|
|
host=host,
|
|
port=port,
|
|
password=password,
|
|
db=db,
|
|
decode_responses=True,
|
|
socket_connect_timeout=2,
|
|
socket_timeout=2
|
|
)
|
|
client.ping()
|
|
_redis_client = client
|
|
log.info(f"✅ Redis conectado en {host}:{port}")
|
|
return _redis_client
|
|
except Exception as e:
|
|
log.warning(f"⚠️ Redis no disponible, usando caché en RAM: {e}")
|
|
return None
|
|
|
|
# ─── API de caché genérica (con TTL en segundos) ───────────────────────────
|
|
|
|
def cache_get(key: str) -> str | None:
|
|
"""Obtiene un valor del caché. Devuelve None si no existe o si Redis no está disponible."""
|
|
r = get_redis()
|
|
if r:
|
|
try:
|
|
return r.get(key)
|
|
except Exception:
|
|
pass
|
|
return None
|
|
|
|
def cache_set(key: str, value: str, ttl: int = 86400) -> None:
|
|
"""Guarda un valor en el caché con TTL en segundos (default 24h)."""
|
|
r = get_redis()
|
|
if r:
|
|
try:
|
|
r.setex(key, ttl, value)
|
|
except Exception:
|
|
pass
|
|
|
|
def cache_delete(key: str) -> None:
|
|
"""Elimina una clave del caché."""
|
|
r = get_redis()
|
|
if r:
|
|
try:
|
|
r.delete(key)
|
|
except Exception:
|
|
pass
|
|
|
|
def cache_increment(key: str, ttl: int = 60) -> int:
|
|
"""Incrementa un contador atómico en Redis. Ideal para Rate Limiting."""
|
|
r = get_redis()
|
|
if r:
|
|
try:
|
|
pipe = r.pipeline()
|
|
pipe.incr(key)
|
|
pipe.expire(key, ttl)
|
|
results = pipe.execute()
|
|
return results[0]
|
|
except Exception:
|
|
pass
|
|
return 0
|