Spaces:
Sleeping
Sleeping
File size: 5,143 Bytes
feeaf83 | 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 | import redis
import json
import logging
import asyncio
from typing import Any, Optional
from datetime import datetime, timedelta
from src.config import settings
logger = logging.getLogger(__name__)
class RedisCacheManager:
def __init__(self):
self.redis_client = None
self.is_active = False
# High-fidelity in-memory cache as backup and local-first store
self.in_memory_fallback = {} # key -> {"data": val, "expires_at": datetime}
self._init_redis()
def _init_redis(self):
if not settings.REDIS_URL:
logger.info("REDIS_URL environment variable is empty. Caching runs in in-memory fallback mode.")
return
try:
# Parse connection URL with robust connection timeout config
self.redis_client = redis.from_url(
settings.REDIS_URL,
decode_responses=True,
socket_connect_timeout=3,
socket_keepalive=True
)
# Test connection
self.redis_client.ping()
self.is_active = True
logger.info("Successfully connected to the remote Redis cache node.")
except Exception as e:
logger.warning(f"Redis connection failed: {e}. Degrading gracefully to in-memory fallback caching.")
self.redis_client = None
self.is_active = False
def _run_sync(self, func, *args, **kwargs):
"""Helper to execute synchronous blocking redis calls in a thread pool executor to avoid blocking FastAPI event loop."""
loop = asyncio.get_event_loop()
return loop.run_in_executor(None, lambda: func(*args, **kwargs))
async def get(self, key: str) -> Optional[Any]:
"""Asynchronously get value from cache with silent graceful fallbacks."""
# 1. If Redis is inactive, check in-memory cache
if not self.is_active or not self.redis_client:
entry = self.in_memory_fallback.get(key)
if entry:
if entry["expires_at"] > datetime.utcnow():
return entry["data"]
else:
del self.in_memory_fallback[key]
return None
# 2. Query Redis via executor
try:
val = await self._run_sync(self.redis_client.get, key)
if val is not None:
try:
return json.loads(val)
except (json.JSONDecodeError, TypeError):
return val
return None
except Exception as e:
logger.error(f"Redis GET failed for key '{key}': {e}. Falling back to local memory.")
# Fall back to local memory check in case connection dropped
entry = self.in_memory_fallback.get(key)
if entry and entry["expires_at"] > datetime.utcnow():
return entry["data"]
return None
async def set(self, key: str, value: Any, ttl: int = 300) -> bool:
"""Asynchronously write to cache with a specified TTL."""
expires_at = datetime.utcnow() + timedelta(seconds=ttl)
# Always maintain the in-memory fallback for local consistency
self.in_memory_fallback[key] = {"data": value, "expires_at": expires_at}
if not self.is_active or not self.redis_client:
return True
try:
serialized = json.dumps(value, default=str)
await self._run_sync(self.redis_client.setex, key, ttl, serialized)
return True
except Exception as e:
logger.error(f"Redis SET failed for key '{key}': {e}. Keeping in local memory.")
return False
async def delete(self, key: str) -> bool:
"""Asynchronously invalidate a key from all caching layers."""
if key in self.in_memory_fallback:
del self.in_memory_fallback[key]
if not self.is_active or not self.redis_client:
return True
try:
await self._run_sync(self.redis_client.delete, key)
return True
except Exception as e:
logger.error(f"Redis DELETE failed for key '{key}': {e}")
return False
async def clear_pattern(self, pattern: str) -> bool:
"""Asynchronously invalidate all keys matching a specific pattern (e.g. 'uniarc:*')."""
# Clear matching local in-memory keys
normalized_pattern = pattern.replace("*", "")
for k in list(self.in_memory_fallback.keys()):
if normalized_pattern in k:
del self.in_memory_fallback[k]
if not self.is_active or not self.redis_client:
return True
try:
keys = await self._run_sync(self.redis_client.keys, pattern)
if keys:
# Delete keys concurrently
await self._run_sync(self.redis_client.delete, *keys)
return True
except Exception as e:
logger.error(f"Redis clear_pattern failed for '{pattern}': {e}")
return False
# Export a single global instance for application-wide sharing
redis_cache = RedisCacheManager()
|