from flask import Flask, jsonify, request, render_template_string import redis import time import os import json app = Flask(__name__) # Connect to Redis (same pod) redis_client = redis.Redis( host='localhost', port=6379, decode_responses=True ) # HTML template for the interface HTML_TEMPLATE = ''' Redis Cache Visualizer

Redis Cache Visualizer

Interactive Redis data management and visualization

📊 Redis Statistics

{{ stats.total_keys }}
Total Keys
{{ stats.used_memory_human }}
Memory Used
{{ stats.connected_clients }}
Connected Clients
{{ stats.uptime_in_seconds // 60 }}
Uptime (min)

🔧 Quick Actions

📝 Add Data

🔍 Search Keys

📋 All Redis Keys

{% for key, value in redis_data.items() %}
{{ key }}: {{ value }}
{% endfor %}

📈 Cache Performance Examples

''' @app.route('/') def index(): """Main dashboard page""" try: # Get Redis statistics info = redis_client.info() # Get all keys (limited for display) keys = redis_client.keys('*') redis_data = {} # Get values for keys (limit to first 20 for display) for key in keys[:20]: try: value = redis_client.get(key) ttl = redis_client.ttl(key) if ttl > 0: redis_data[key] = f"{value} (TTL: {ttl}s)" else: redis_data[key] = value except: redis_data[key] = "[Binary Data]" # Format statistics for display stats = { 'total_keys': len(keys), 'used_memory_human': info.get('used_memory_human', 'N/A'), 'connected_clients': info.get('connected_clients', 'N/A'), 'uptime_in_seconds': info.get('uptime_in_seconds', 'N/A'), 'total_commands_processed': info.get('total_commands_processed', 'N/A'), 'keyspace_hits': info.get('keyspace_hits', 'N/A'), 'keyspace_misses': info.get('keyspace_misses', 'N/A'), 'expired_keys': info.get('expired_keys', 'N/A'), 'evicted_keys': info.get('evicted_keys', 'N/A') } return render_template_string(HTML_TEMPLATE, redis_data=redis_data, stats=stats) except Exception as e: return f"

Error connecting to Redis: {e}

" @app.route('/api/stats') def get_stats(): """API endpoint for Redis statistics""" try: info = redis_client.info() return jsonify({ 'status': 'success', 'data': { 'keys': len(redis_client.keys('*')), 'memory': info.get('used_memory_human'), 'clients': info.get('connected_clients'), 'uptime_minutes': info.get('uptime_in_seconds', 0) // 60 } }) except Exception as e: return jsonify({'status': 'error', 'message': str(e)}), 500 @app.route('/api/keys') def get_keys(): """API endpoint to get all keys""" try: keys = redis_client.keys('*') return jsonify({'status': 'success', 'keys': keys}) except Exception as e: return jsonify({'status': 'error', 'message': str(e)}), 500 @app.route('/api/set', methods=['POST']) def set_key(): """API endpoint to set a key-value pair""" try: data = request.json key = data.get('key') value = data.get('value') ttl = data.get('ttl') if not key or value is None: return jsonify({'status': 'error', 'message': 'Key and value are required'}), 400 if ttl and ttl > 0: redis_client.setex(key, ttl, value) else: redis_client.set(key, value) return jsonify({'status': 'success', 'message': f'Key {key} set successfully'}) except Exception as e: return jsonify({'status': 'error', 'message': str(e)}), 500 @app.route('/api/get/') def get_key(key): """API endpoint to get a key-value pair""" try: value = redis_client.get(key) ttl = redis_client.ttl(key) if value is None: return jsonify({'status': 'error', 'message': 'Key not found'}), 404 return jsonify({ 'status': 'success', 'data': { 'key': key, 'value': value, 'ttl': ttl } }) except Exception as e: return jsonify({'status': 'error', 'message': str(e)}), 500 @app.route('/api/delete/', methods=['DELETE']) def delete_key(key): """API endpoint to delete a key""" try: deleted = redis_client.delete(key) if deleted == 0: return jsonify({'status': 'error', 'message': 'Key not found'}), 404 return jsonify({'status': 'success', 'message': f'Key {key} deleted successfully'}) except Exception as e: return jsonify({'status': 'error', 'message': str(e)}), 500 @app.route('/api/search/') def search_keys(pattern): """API endpoint to search for keys""" try: keys = redis_client.keys(pattern) return jsonify({'status': 'success', 'keys': keys}) except Exception as e: return jsonify({'status': 'error', 'message': str(e)}), 500 @app.route('/api/add_sample', methods=['POST']) def add_sample_data(): """Add sample data for demonstration""" try: # Add various types of data redis_client.set('user:1001', 'Alice Johnson') redis_client.set('user:1002', 'Bob Smith') redis_client.set('session:abc123', 'logged_in') redis_client.setex('temp:token', 300, 'temporary_data') redis_client.set('api:request:count', '0') redis_client.set('cache:homepage', 'Cached homepage content') redis_client.setex('user:login:last', 3600, '2025-11-04T17:00:00') return jsonify({'status': 'success', 'message': 'Sample data added'}) except Exception as e: return jsonify({'status': 'error', 'message': str(e)}), 500 @app.route('/api/clear_all', methods=['POST']) def clear_all(): """Clear all Redis data""" try: redis_client.flushall() return jsonify({'status': 'success', 'message': 'All data cleared'}) except Exception as e: return jsonify({'status': 'error', 'message': str(e)}), 500 @app.route('/api/demo_cache', methods=['POST']) def demo_cache(): """Demonstrate caching patterns""" try: # Simulate cache-aside pattern redis_client.setex('api:users:1001', 600, 'User data from database') redis_client.setex('api:posts:recent', 300, 'Recent blog posts') redis_client.setex('api:stats:daily', 1800, 'Daily statistics') return jsonify({'status': 'success', 'message': 'Cache examples added'}) except Exception as e: return jsonify({'status': 'error', 'message': str(e)}), 500 @app.route('/api/demo_session', methods=['POST']) def demo_session(): """Demonstrate session storage""" try: # Simulate session storage redis_client.setex('session:user123', 3600, 'active') redis_client.setex('session:user456', 7200, 'premium') redis_client.setex('session:admin789', 1800, 'administrator') return jsonify({'status': 'success', 'message': 'Session examples added'}) except Exception as e: return jsonify({'status': 'error', 'message': str(e)}), 500 @app.route('/api/demo_counter', methods=['POST']) def demo_counter(): """Demonstrate counter/rate limiting""" try: # Simulate counters redis_client.set('counter:api_requests', '0') redis_client.set('counter:user_visits', '0') redis_client.set('counter:page_views', '0') return jsonify({'status': 'success', 'message': 'Counter examples added'}) except Exception as e: return jsonify({'status': 'error', 'message': str(e)}), 500 @app.route('/api/memory_info') def memory_info(): """Get detailed memory information""" try: info = redis_client.info('memory') return jsonify({ 'status': 'success', 'memory': { 'used_memory': info.get('used_memory'), 'used_memory_human': info.get('used_memory_human'), 'used_memory_rss': info.get('used_memory_rss'), 'maxmemory': info.get('maxmemory'), 'maxmemory_policy': info.get('maxmemory_policy'), 'mem_fragmentation_ratio': info.get('mem_fragmentation_ratio') } }) except Exception as e: return jsonify({'status': 'error', 'message': str(e)}), 500 if __name__ == '__main__': print("Starting Redis Web Interface...") print("Access Redis at: localhost:6379") print("Access Web Interface at: localhost:8080") # Test Redis connection try: redis_client.ping() print("✅ Redis connection successful") except Exception as e: print(f"❌ Redis connection failed: {e}") exit(1) app.run(host='0.0.0.0', port=8080, debug=False)