"""Cache management API endpoints.""" import logging from fastapi import APIRouter, HTTPException from typing import Dict, Any from app.services import cache logger = logging.getLogger(__name__) router = APIRouter(prefix="/api/v1/cache", tags=["Cache Management"]) @router.get("/stats", response_model=Dict[str, Any]) async def get_cache_stats(): """ Get cache statistics. Returns: - hits: Number of cache hits - misses: Number of cache misses - sets: Number of cache writes - total_keys: Current number of cached route keys - enabled: Whether Redis cache is enabled """ try: stats = cache.get_stats() # Calculate hit rate total_requests = stats.get("hits", 0) + stats.get("misses", 0) if total_requests > 0: stats["hit_rate"] = round(stats.get("hits", 0) / total_requests * 100, 2) else: stats["hit_rate"] = 0.0 return stats except Exception as e: logger.error(f"Error getting cache stats: {e}") raise HTTPException(status_code=500, detail="Internal server error") @router.get("/keys") async def list_cache_keys(pattern: str = "routes:*"): """ List cache keys matching pattern. - **pattern**: Redis key pattern (default: "routes:*") """ try: keys = cache.get_keys(pattern) return { "pattern": pattern, "count": len(keys), "keys": keys[:100] # Limit to first 100 for response size } except Exception as e: logger.error(f"Error listing cache keys: {e}") raise HTTPException(status_code=500, detail="Internal server error") @router.delete("/clear") async def clear_cache(pattern: str = "routes:*"): """ Clear cache keys matching pattern. - **pattern**: Redis key pattern to delete (default: "routes:*") [WARN] **Warning**: This will delete cached route optimizations! """ try: deleted_count = cache.delete(pattern) logger.info(f"Cleared {deleted_count} cache keys matching pattern: {pattern}") return { "pattern": pattern, "deleted_count": deleted_count, "message": f"Cleared {deleted_count} cache keys" } except Exception as e: logger.error(f"Error clearing cache: {e}") raise HTTPException(status_code=500, detail="Internal server error")