Spaces:
Running
Running
File size: 2,363 Bytes
565a379 | 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 | import os
import json
import logging
from typing import Optional, Any, Union
import redis
from redis.exceptions import RedisError
# Configure logging
logger = logging.getLogger(__name__)
class RedisCache:
def __init__(self, redis_url: Optional[str] = None):
"""
Initialize the RedisCache wrapper.
Args:
redis_url: Connection URL for Redis. Defaults to REDIS_URL env var or localhost.
"""
self.redis_url = redis_url or os.getenv("REDIS_URL", "redis://localhost:6379/0")
self.client = None
self._connect()
def _connect(self):
"""Establish Redis connection."""
try:
self.client = redis.from_url(self.redis_url, decode_responses=True)
self.client.ping()
logger.info("Connected to Redis cache.")
except RedisError as e:
logger.error(f"Failed to connect to Redis: {e}")
self.client = None
def get(self, problem_hash: str) -> Optional[Any]:
"""
Retrieve result from cache using problem_hash.
Args:
problem_hash: The unique hash of the problem.
Returns:
The cached result (deserialized) or None if not found/error.
"""
if not self.client:
return None
try:
data = self.client.get(problem_hash)
if data:
return json.loads(data)
return None
except (RedisError, json.JSONDecodeError) as e:
logger.error(f"Error retrieving from cache: {e}")
return None
def set(self, problem_hash: str, result: Any, ttl: int = 86400) -> bool:
"""
Store result in cache with TTL.
Args:
problem_hash: The unique hash key.
result: The data to cache (will be JSON serialized).
ttl: Time-to-live in seconds (default 1 day).
Returns:
True if successful, False otherwise.
"""
if not self.client:
return False
try:
serialized = json.dumps(result)
return bool(self.client.setex(problem_hash, ttl, serialized))
except (RedisError, TypeError) as e:
logger.error(f"Error setting cache: {e}")
return False
|