"""Redis client module for caching and storage.""" import json import os from typing import Optional, Any import redis.asyncio as redis from functools import wraps import hashlib # Redis configuration from environment variables REDIS_HOST = os.getenv("REDIS_HOST", "localhost") REDIS_PORT = int(os.getenv("REDIS_PORT", 6379)) REDIS_DB = int(os.getenv("REDIS_DB", 0)) REDIS_PASSWORD = os.getenv("REDIS_PASSWORD", None) CACHE_TTL = int(os.getenv("CACHE_TTL", 300)) # Default 5 minutes # Global Redis client instance _redis_client: Optional[redis.Redis] = None def get_redis_client() -> redis.Redis: """Get or create the Redis client instance.""" global _redis_client if _redis_client is None: _redis_client = redis.Redis( host=REDIS_HOST, port=REDIS_PORT, db=REDIS_DB, password=REDIS_PASSWORD, decode_responses=True, ) return _redis_client async def init_redis() -> None: """Initialize Redis connection.""" global _redis_client _redis_client = redis.Redis( host=REDIS_HOST, port=REDIS_PORT, db=REDIS_DB, password=REDIS_PASSWORD, decode_responses=True, ) try: await _redis_client.ping() print(f"✓ Connected to Redis at {REDIS_HOST}:{REDIS_PORT}") except redis.ConnectionError as e: print(f"⚠ Redis connection failed: {e}") print("⚠ Running without Redis caching") async def close_redis() -> None: """Close Redis connection.""" global _redis_client if _redis_client: await _redis_client.close() _redis_client = None def _generate_cache_key(prefix: str, *args, **kwargs) -> str: """Generate a cache key from arguments.""" key_parts = [prefix] for arg in args: key_parts.append(str(arg)) for k, v in sorted(kwargs.items()): key_parts.append(f"{k}={v}") key_string = ":".join(key_parts) return f"cache:{hashlib.md5(key_string.encode()).hexdigest()}" def cache_it(prefix: str, ttl: int = CACHE_TTL): """Decorator to cache function results in Redis. Args: prefix: Key prefix for the cache ttl: Time to live in seconds """ def decorator(func): @wraps(func) async def wrapper(*args, **kwargs): client = get_redis_client() cache_key = _generate_cache_key(prefix, *args, **kwargs) # Try to get from cache try: cached = await client.get(cache_key) if cached: return json.loads(cached) except Exception: pass # Cache miss or error, proceed to fetch # Fetch fresh data result = func(*args, **kwargs) # Store in cache try: await client.setex(cache_key, ttl, json.dumps(result)) except Exception: pass # Cache write failed, but return the result anyway return result return wrapper return decorator # ============== Cache helper functions ============== async def cache_get(key: str) -> Optional[Any]: """Get a value from cache.""" try: client = get_redis_client() value = await client.get(key) return json.loads(value) if value else None except Exception: return None async def cache_set(key: str, value: Any, ttl: int = CACHE_TTL) -> bool: """Set a value in cache.""" try: client = get_redis_client() await client.setex(key, ttl, json.dumps(value)) return True except Exception: return False async def cache_delete(key: str) -> bool: """Delete a key from cache.""" try: client = get_redis_client() await client.delete(key) return True except Exception: return False async def cache_delete_pattern(pattern: str) -> bool: """Delete keys matching a pattern.""" try: client = get_redis_client() keys = [] async for key in client.scan_iter(match=pattern): keys.append(key) if keys: await client.delete(*keys) return True except Exception: return False # ============== Entity-specific cache functions ============== # Songs cache async def cache_song(song_id: int, song_data: dict) -> bool: """Cache a song by ID.""" return await cache_set(f"song:{song_id}", song_data) async def get_cached_song(song_id: int) -> Optional[dict]: """Get a cached song by ID.""" return await cache_get(f"song:{song_id}") async def invalidate_song_cache(song_id: int) -> None: """Invalidate song cache.""" await cache_delete(f"song:{song_id}") # Users cache async def cache_user(user_id: int, user_data: dict) -> bool: """Cache a user by ID.""" return await cache_set(f"user:{user_id}", user_data) async def get_cached_user(user_id: int) -> Optional[dict]: """Get a cached user by ID.""" return await cache_get(f"user:{user_id}") async def invalidate_user_cache(user_id: int) -> None: """Invalidate user cache.""" await cache_delete(f"user:{user_id}") # Playlists cache async def cache_playlist(playlist_id: int, playlist_data: dict) -> bool: """Cache a playlist by ID.""" return await cache_set(f"playlist:{playlist_id}", playlist_data) async def get_cached_playlist(playlist_id: int) -> Optional[dict]: """Get a cached playlist by ID.""" return await cache_get(f"playlist:{playlist_id}") async def invalidate_playlist_cache(playlist_id: int) -> None: """Invalidate playlist cache.""" await cache_delete(f"playlist:{playlist_id}") # Lists cache (for collections) async def cache_list(key: str, data: list) -> bool: """Cache a list of items.""" return await cache_set(f"list:{key}", data, ttl=60) # Shorter TTL for lists async def get_cached_list(key: str) -> Optional[list]: """Get a cached list.""" return await cache_get(f"list:{key}") async def invalidate_list_cache(key: str) -> None: """Invalidate a list cache.""" await cache_delete(f"list:{key}") # ============== Play History in Redis (hot storage) ============== async def store_play_event(user_id: int, song_id: int, context_id: Optional[int] = None) -> dict: """Store a play event in Redis (fast write).""" from datetime import datetime played_at = datetime.utcnow().isoformat() event = { "user_id": user_id, "song_id": song_id, "context_id": context_id, "played_at": played_at } try: client = get_redis_client() # Add to sorted set for this user (score = timestamp) await client.zadd( f"play_history:user:{user_id}", {json.dumps(event): datetime.utcnow().timestamp()} ) # Also add to global history await client.zadd( "play_history:global", {json.dumps(event): datetime.utcnow().timestamp()} ) except Exception: pass # Best effort, SQLite is the source of truth return event async def get_user_play_history(user_id: int, limit: int = 50) -> list[dict]: """Get play history for a user from Redis.""" try: client = get_redis_client() # Get latest entries from sorted set events = await client.zrevrange( f"play_history:user:{user_id}", 0, limit - 1 ) return [json.loads(e) for e in events] except Exception: return [] async def get_global_play_history(limit: int = 50) -> list[dict]: """Get global play history from Redis.""" try: client = get_redis_client() events = await client.zrevrange( "play_history:global", 0, limit - 1 ) return [json.loads(e) for e in events] except Exception: return []