65 lines
2.3 KiB
Python
65 lines
2.3 KiB
Python
import os
|
|
import sys
|
|
import redis
|
|
import argparse
|
|
from dotenv import load_dotenv
|
|
|
|
# Safely load environment variables
|
|
base_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
env_path = os.path.join(base_dir, 'app', 'config', '.env')
|
|
|
|
load_dotenv(env_path)
|
|
|
|
def flush_redis():
|
|
parser = argparse.ArgumentParser(description="Safely flush Redis DB 0")
|
|
parser.add_argument("--confirm", action="store_true", help="Explicitly confirm the wipe")
|
|
parser.add_argument("--force", action="store_true", help="Force wipe even in production")
|
|
parser.add_argument("--dry-run", action="store_true", help="Check connection without flushing")
|
|
|
|
args = parser.parse_args()
|
|
|
|
# Environment safety check
|
|
app_env = os.getenv('APP_ENV', 'development').lower()
|
|
is_production = app_env in ['production', 'prod', 'staging']
|
|
|
|
redis_host = os.getenv('REDIS_HOST', 'localhost')
|
|
redis_port = os.getenv('REDIS_PORT', '6379')
|
|
redis_password = os.getenv('REDIS_PASSWORD')
|
|
|
|
if is_production and not args.force:
|
|
print(f"❌ CRITICAL: Detected PRODUCTION environment ({app_env}).")
|
|
print(" Refusing to flush without the --force flag.")
|
|
sys.exit(1)
|
|
|
|
if not args.confirm and not args.dry_run:
|
|
print("⚠️ WARNING: You are about to IRRECOVERABLY WIPE Redis DB 0.")
|
|
print(" This will clear sessions, task queues, and blacklisted tokens.")
|
|
confirmation = input(" To proceed, type 'CONFIRM': ")
|
|
if confirmation != "CONFIRM":
|
|
print("❌ Aborted. No changes made.")
|
|
sys.exit(0)
|
|
|
|
if redis_password:
|
|
redis_url = f"redis://:{redis_password}@{redis_host}:{redis_port}/0"
|
|
else:
|
|
redis_url = f"redis://{redis_host}:{redis_port}/0"
|
|
|
|
print(f"Connecting to Redis at {redis_host}:{redis_port}...")
|
|
try:
|
|
client = redis.Redis.from_url(redis_url)
|
|
|
|
if args.dry_run:
|
|
client.ping()
|
|
print("✅ Dry-run: Connection successful. No data was deleted.")
|
|
return
|
|
|
|
print(f"Flushing DB 0 ({app_env})...")
|
|
client.flushdb()
|
|
print("✅ Successfully flushed Redis DB 0")
|
|
except Exception as e:
|
|
print(f"❌ Error flushing Redis: {e}")
|
|
sys.exit(1)
|
|
|
|
if __name__ == "__main__":
|
|
flush_redis()
|