Spaces:
Paused
Paused
File size: 5,879 Bytes
83bdb4a 89e2f2e 83bdb4a 89e2f2e 83bdb4a 89e2f2e 83bdb4a 89e2f2e 83bdb4a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 | """
Redis Client Pool — CrowData
Centralized Redis connection management with in-memory LRU fallback.
"""
import asyncio
import json
import logging
import time
from collections import OrderedDict
from typing import Optional
try:
import redis.asyncio as aioredis
_REDIS_MODULE_AVAILABLE = True
except ImportError:
aioredis = None
_REDIS_MODULE_AVAILABLE = False
from app.config import get_settings
logger = logging.getLogger(__name__)
settings = get_settings()
# Global connection pools
_redis_client = None
_redis_is_available = True
_reconnect_task = None
# In-memory LRU cache fallback
class LRUCache:
def __init__(self, maxsize: int = 5000):
self.maxsize = maxsize
self._cache = OrderedDict()
self._ttls = {}
self._lock = asyncio.Lock()
async def get(self, key: str):
async with self._lock:
if key not in self._cache:
return None
expire_at = self._ttls.get(key, 0)
if expire_at and expire_at < time.time():
# Expired
self._cache.pop(key, None)
self._ttls.pop(key, None)
return None
# Move to end (most recently used)
value = self._cache.pop(key)
self._cache[key] = value
return value
async def set(self, key: str, value: dict, ttl: int):
async with self._lock:
# Evict if at maxsize
if len(self._cache) >= self.maxsize and key not in self._cache:
self._cache.popitem(last=False) # Remove LRU
self._cache[key] = value
self._ttls[key] = time.time() + ttl if ttl > 0 else 0
async def delete(self, key: str):
async with self._lock:
self._cache.pop(key, None)
self._ttls.pop(key, None)
async def cleanup_expired(self):
"""Remove expired entries. Called periodically."""
async with self._lock:
now = time.time()
expired = [k for k, exp in self._ttls.items() if exp and exp < now]
for k in expired:
self._cache.pop(k, None)
self._ttls.pop(k, None)
def __len__(self):
return len(self._cache)
_in_memory_cache = LRUCache(maxsize=5000)
_redis_client = None
_redis_is_available = _REDIS_MODULE_AVAILABLE # False si el módulo no está instalado
_reconnect_task = None
async def get_redis():
global _redis_client
if not _REDIS_MODULE_AVAILABLE:
raise RuntimeError("redis module not installed; using in-memory fallback")
if _redis_client is None:
from app.config import get_settings
settings = get_settings()
_redis_client = await aioredis.from_url(
settings.redis_url,
encoding="utf-8",
decode_responses=True,
max_connections=getattr(settings, 'redis_max_connections', 20) or 20,
socket_keepalive=True,
socket_connect_timeout=5,
socket_timeout=5,
retry_on_timeout=True,
)
return _redis_client
async def _try_reconnect():
"""Background task to attempt Redis reconnection."""
global _redis_is_available, _reconnect_task
while True:
await asyncio.sleep(30) # Try every 30 seconds
if not _redis_is_available:
try:
r = await get_redis()
await r.ping()
_redis_is_available = True
logger.info("Redis reconnected successfully")
except Exception:
pass
async def _start_reconnect_task():
global _reconnect_task
if _reconnect_task is None or _reconnect_task.done():
_reconnect_task = asyncio.create_task(_try_reconnect())
async def cache_get(key: str):
global _redis_is_available
if _redis_is_available:
try:
r = await get_redis()
data = await r.get(key)
if data:
return json.loads(data)
except Exception as e:
logger.warning(f"Redis GET error for key {key}: {e}. Falling back to in-memory cache.")
_redis_is_available = False
# Start background reconnection
asyncio.create_task(_try_reconnect())
# In-memory fallback
return await _in_memory_cache.get(key)
async def cache_set(key: str, value: dict, ttl: int = None):
global _redis_is_available
ttl = ttl or 86400
if _redis_is_available:
try:
r = await get_redis()
await r.setex(key, ttl, json.dumps(value, ensure_ascii=False))
return
except Exception as e:
logger.warning(f"Redis SET error for key {key}: {e}. Saving in memory.")
_redis_is_available = False
# Start background reconnection
asyncio.create_task(_try_reconnect())
# In-memory LRU fallback
await _in_memory_cache.set(key, value, ttl=60)
async def cache_delete(key: str):
global _redis_is_available
if _redis_is_available:
try:
r = await get_redis()
await r.delete(key)
except Exception as e:
logger.warning(f"Redis DELETE error for key {key}: {e}.")
_redis_is_available = False
asyncio.create_task(_try_reconnect())
# In-memory fallback
await _in_memory_cache.delete(key)
async def cleanup_expired():
"""Periodic cleanup of expired in-memory cache entries."""
await _in_memory_cache.cleanup_expired()
async def _start_reconnect_task():
"""Start the background reconnection task."""
pass # Task is already started at module level
# Module-level initialization - start background tasks
try:
loop = asyncio.get_running_loop()
loop.create_task(_try_reconnect())
except RuntimeError:
# No event loop running, will be created later
pass |