File size: 4,854 Bytes
2ed8996 | 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 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 | """
SQLite Fallback Database Setup and Test Script
This script sets up and tests the SQLite fallback database system
for the AegisLM backend application.
"""
import asyncio
import logging
import sys
from pathlib import Path
# Add the backend directory to Python path
backend_dir = Path(__file__).parent
sys.path.insert(0, str(backend_dir))
from core.config import settings
from core.sqlite_fallback_manager import SQLiteFallbackManager, setup_sqlite_fallback, test_fallback_mechanism
# Setup logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
async def main():
"""Main function to setup and test SQLite fallback."""
print("=" * 60)
print("SQLite Fallback Database Setup and Test")
print("=" * 60)
# Check configuration
print(f"\n📋 Configuration:")
print(f" SQLite Fallback Enabled: {settings.ENABLE_SQLITE_FALLBACK}")
print(f" SQLite Database Path: {settings.SQLITE_DATABASE_PATH}")
print(f" Fallback Timeout: {settings.SQLITE_FALLBACK_TIMEOUT} seconds")
if not settings.ENABLE_SQLITE_FALLBACK:
print("\n❌ SQLite fallback is disabled in configuration")
print(" Set ENABLE_SQLITE_FALLBACK=true in .env to enable")
return
try:
# Step 1: Setup SQLite fallback
print("\n🚀 Setting up SQLite fallback database...")
setup_success = await setup_sqlite_fallback()
if setup_success:
print("✅ SQLite fallback setup completed successfully")
else:
print("❌ SQLite fallback setup failed")
return
# Step 2: Get status
print("\n📊 Getting fallback database status...")
manager = SQLiteFallbackManager()
status = await manager.get_fallback_status()
print("Status Information:")
for key, value in status.items():
if key == "fallback_database_modified" and value:
import datetime
modified_time = datetime.datetime.fromtimestamp(value)
print(f" {key}: {modified_time}")
else:
print(f" {key}: {value}")
# Step 3: Test fallback mechanism
print("\n🧪 Testing fallback mechanism...")
test_results = await test_fallback_mechanism()
print("Test Results:")
for key, value in test_results.items():
if key == "tests":
print(" Individual Tests:")
for test_name, result in value.items():
status_icon = "✅" if "PASS" in result else "❌" if "FAIL" in result else "⏭️"
print(f" {status_icon} {test_name}: {result}")
else:
status_icon = "✅" if value else "❌"
print(f" {status_icon} {key}: {value}")
# Step 4: Check database file
db_path = Path(settings.SQLITE_DATABASE_PATH)
print(f"\n📁 Database File Information:")
print(f" Path: {db_path.absolute()}")
print(f" Exists: {db_path.exists()}")
if db_path.exists():
size_mb = db_path.stat().st_size / (1024 * 1024)
print(f" Size: {size_mb:.2f} MB")
# Check if database is readable
try:
import sqlite3
conn = sqlite3.connect(str(db_path))
cursor = conn.cursor()
cursor.execute("SELECT name FROM sqlite_master WHERE type='table'")
tables = cursor.fetchall()
conn.close()
print(f" Tables: {len(tables)}")
for table_name in tables:
print(f" - {table_name[0]}")
except Exception as e:
print(f" Error reading database: {e}")
print("\n" + "=" * 60)
print("SQLite Fallback Database Setup and Test - COMPLETED")
print("=" * 60)
# Summary
if test_results.get("primary_health") and test_results.get("fallback_health"):
print("🎉 Both PostgreSQL and SQLite databases are healthy!")
print("🔄 Automatic fallback is ready and functional.")
elif test_results.get("fallback_health"):
print("✅ SQLite fallback database is healthy and ready!")
print("⚠️ PostgreSQL is unavailable, fallback will be used.")
else:
print("❌ Both databases are unavailable - check configuration")
except Exception as e:
logger.error(f"Setup and test failed: {e}")
print(f"\n❌ Error: {e}")
print(" Check the logs above for detailed error information")
if __name__ == "__main__":
asyncio.run(main())
|