Spaces:
Runtime error
Runtime error
File size: 2,368 Bytes
ac90985 | 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 | import asyncio
import logging
from app.core.config import settings
from app.services.cache import cache_service
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
async def clear_all_cache():
"""
Clears all cache from Redis and local memory.
"""
logger.info("🧹 Starting cache clearing process...")
# 1. Clear Redis Cache
if hasattr(cache_service, 'redis_client') and cache_service.redis_client:
try:
cache_service.redis_client.flushdb()
logger.info("✅ Redis cache flushed successfully.")
except Exception as e:
logger.error(f"❌ Failed to flush Redis: {e}")
elif hasattr(cache_service, 'redis') and cache_service.redis:
# In case the attribute name is just 'redis' (checking implementation details)
try:
cache_service.redis.flushdb()
logger.info("✅ Redis cache flushed successfully.")
except Exception as e:
logger.error(f"❌ Failed to flush Redis: {e}")
else:
# Check if REDIS_AVAILABLE global is true in the module, but here we work with the instance
# The instance uses 'redis_client' variable in the module scope, but doesn't expose it as self.redis_client
# Wait, looking at the code:
# redis_client is a global variable in the module, not a class attribute of CacheService instance (unless assigned).
# But wait, looking at lines 27/39/57 of cache.py, it uses the global `redis_client`.
# To access it from outside, we need to import it or usage method.
# Use direct redis connection to be sure.
try:
import redis
r = redis.from_url(settings.REDIS_URL)
r.flushdb()
logger.info("✅ Redis cache flushed via direct connection.")
except Exception as e:
logger.warning(f"⚠️ Could not connect/flush Redis directly (might not be configured): {e}")
# 2. Clear Local Cache
if hasattr(cache_service, 'local_cache'):
count = len(cache_service.local_cache)
cache_service.local_cache.clear()
logger.info(f"✅ Local dictionary cache cleared ({count} items removed).")
logger.info("✨ System cache cleared successfully.")
if __name__ == "__main__":
asyncio.run(clear_all_cache())
|