55 lines
1.8 KiB
Python
55 lines
1.8 KiB
Python
import time
|
|
from datetime import datetime
|
|
from app.database import db_manager, CACHE_EXPIRY_HOURS
|
|
|
|
def clear_cache():
|
|
"""Clear the entire cache database"""
|
|
conn = db_manager.get_connection()
|
|
cursor = conn.cursor()
|
|
cursor.execute("DELETE FROM cache")
|
|
conn.commit()
|
|
conn.close()
|
|
return {"status": "success", "message": "Cache cleared successfully"}
|
|
|
|
def get_cache_stats():
|
|
"""Get cache statistics"""
|
|
conn = db_manager.get_connection()
|
|
cursor = conn.cursor()
|
|
|
|
# Get total entries
|
|
cursor.execute("SELECT COUNT(*) FROM cache")
|
|
total_entries = cursor.fetchone()[0]
|
|
|
|
# Get entries by route
|
|
cursor.execute("SELECT route, COUNT(*) FROM cache GROUP BY route")
|
|
routes = {route: count for route, count in cursor.fetchall()}
|
|
|
|
# Get recent entries (last 24 hours)
|
|
recent_timestamp = int(time.time()) - (24 * 60 * 60)
|
|
cursor.execute("SELECT COUNT(*) FROM cache WHERE timestamp > %s", (recent_timestamp,))
|
|
recent_entries = cursor.fetchone()[0]
|
|
|
|
# Get oldest entry timestamp
|
|
cursor.execute("SELECT MIN(timestamp) FROM cache")
|
|
oldest_timestamp = cursor.fetchone()[0]
|
|
oldest_date = datetime.fromtimestamp(oldest_timestamp).isoformat() if oldest_timestamp else None
|
|
|
|
# Get newest entry timestamp
|
|
cursor.execute("SELECT MAX(timestamp) FROM cache")
|
|
newest_timestamp = cursor.fetchone()[0]
|
|
newest_date = datetime.fromtimestamp(newest_timestamp).isoformat() if newest_timestamp else None
|
|
|
|
conn.close()
|
|
|
|
return {
|
|
"status": "success",
|
|
"stats": {
|
|
"total_entries": total_entries,
|
|
"entries_by_route": routes,
|
|
"recent_entries": recent_entries,
|
|
"oldest_entry": oldest_date,
|
|
"newest_entry": newest_date,
|
|
"cache_expiry_hours": CACHE_EXPIRY_HOURS,
|
|
"database_type": db_manager.db_type
|
|
}
|
|
} |