| """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_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)) |
|
|
| |
| _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: |
| cached = await client.get(cache_key) |
| if cached: |
| return json.loads(cached) |
| except Exception: |
| pass |
| |
| |
| result = func(*args, **kwargs) |
| |
| |
| try: |
| await client.setex(cache_key, ttl, json.dumps(result)) |
| except Exception: |
| pass |
| |
| return result |
| return wrapper |
| return decorator |
|
|
|
|
| |
|
|
| 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 |
|
|
|
|
| |
|
|
| |
| 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}") |
|
|
|
|
| |
| 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}") |
|
|
|
|
| |
| 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}") |
|
|
|
|
| |
| async def cache_list(key: str, data: list) -> bool: |
| """Cache a list of items.""" |
| return await cache_set(f"list:{key}", data, ttl=60) |
|
|
|
|
| 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}") |
|
|
|
|
| |
|
|
| 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() |
| |
| await client.zadd( |
| f"play_history:user:{user_id}", |
| {json.dumps(event): datetime.utcnow().timestamp()} |
| ) |
| |
| await client.zadd( |
| "play_history:global", |
| {json.dumps(event): datetime.utcnow().timestamp()} |
| ) |
| except Exception: |
| pass |
| |
| 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() |
| |
| 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 [] |
|
|
