File size: 4,381 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 | """
Simple SQLite Schema Verification
This script checks if the SQLite fallback database has the required tables.
"""
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 sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import text
# Setup logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
async def check_sqlite_schema():
"""Check SQLite database schema."""
print("🔍 SQLite Fallback Database Schema Check")
print("=" * 50)
try:
# Import database components
from core.config import settings
from core.fallback_database import check_fallback_health, get_fallback_db
print(f"SQLite Fallback Enabled: {settings.ENABLE_SQLITE_FALLBACK}")
print(f"SQLite Database Path: {settings.SQLITE_DATABASE_PATH}")
# Check if fallback is available
if not await check_fallback_health():
print("❌ SQLite fallback database is not available")
return False
print("✅ SQLite fallback database is available")
# Get tables from SQLite
async for session in get_fallback_db():
result = await session.execute(text("""
SELECT name FROM sqlite_master
WHERE type='table'
AND name NOT LIKE 'sqlite_%'
ORDER BY name
"""))
tables = result.fetchall()
table_names = [table[0] for table in tables]
print(f"\n📋 Found {len(table_names)} tables in SQLite:")
for table_name in table_names:
print(f" - {table_name}")
break
# Expected tables based on models
expected_tables = {
'users', 'evaluations', 'api_keys', 'roles', 'permissions',
'user_roles' # Junction table for many-to-many relationship
}
print(f"\n📊 Expected tables: {len(expected_tables)}")
for table in expected_tables:
status = "✅" if table in table_names else "❌"
print(f" {status} {table}")
# Check for missing tables
missing_tables = expected_tables - set(table_names)
extra_tables = set(table_names) - expected_tables
if missing_tables:
print(f"\n❌ Missing tables: {missing_tables}")
return False
else:
print(f"\n✅ All required tables present!")
if extra_tables:
print(f"ℹ️ Extra tables: {extra_tables}")
# Get detailed schema for each table
print(f"\n🔍 Detailed Schema:")
for table_name in table_names:
if table_name in expected_tables:
print(f"\n📝 Table: {table_name}")
# Get columns
columns_result = await session.execute(text(f"PRAGMA table_info({table_name})"))
columns = columns_result.fetchall()
for col in columns:
nullable = "NULL" if col[3] == 1 else "NOT NULL"
pk = " (PK)" if col[5] == 1 else ""
print(f" - {col[1]}: {col[2]} {nullable}{pk}")
return True
except Exception as e:
print(f"❌ Error checking schema: {e}")
logger.error(f"Schema check failed: {e}")
return False
async def main():
"""Main function."""
success = await check_sqlite_schema()
print("\n" + "=" * 50)
if success:
print("🎉 SQLite schema verification PASSED")
print(" All required tables are present in the fallback database!")
else:
print("❌ SQLite schema verification FAILED")
print(" Some issues were detected with the fallback database.")
print("=" * 50)
return 0 if success else 1
if __name__ == "__main__":
exit_code = asyncio.run(main())
exit(exit_code)
|