279 lines
11 KiB
Python
279 lines
11 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Comprehensive Database Schema Analysis and Fix
|
||
Analyzes and fixes all database issues:
|
||
- Schema correctness
|
||
- Missing columns
|
||
- Foreign key constraints
|
||
- Indexes for performance
|
||
- Backend/database alignment
|
||
"""
|
||
|
||
from postgresql_models import engine, Base, Profile, Song, Plan, PlanSong, ProfileSong, ProfileSongKey
|
||
from sqlalchemy import text, inspect, Column, String, Text
|
||
import sys
|
||
|
||
def analyze_schema():
|
||
"""Analyze current database schema and identify issues"""
|
||
print("=" * 70)
|
||
print("DATABASE SCHEMA ANALYSIS")
|
||
print("=" * 70)
|
||
|
||
inspector = inspect(engine)
|
||
issues = []
|
||
fixes = []
|
||
|
||
# Check Profile table
|
||
print("\n📋 CHECKING PROFILES TABLE")
|
||
print("-" * 70)
|
||
profile_cols = {col['name']: col for col in inspector.get_columns('profiles')}
|
||
|
||
# Check for missing columns referenced in app.py
|
||
required_profile_cols = {
|
||
'email': 'VARCHAR(255)',
|
||
'contact_number': 'VARCHAR(50)',
|
||
'notes': 'TEXT'
|
||
}
|
||
|
||
for col_name, col_type in required_profile_cols.items():
|
||
if col_name not in profile_cols:
|
||
issues.append(f"❌ Profile.{col_name} column is MISSING (referenced in app.py)")
|
||
fixes.append(f"ALTER TABLE profiles ADD COLUMN {col_name} {col_type} DEFAULT '';")
|
||
else:
|
||
print(f" ✅ Profile.{col_name} exists")
|
||
|
||
# Check Songs table
|
||
print("\n📋 CHECKING SONGS TABLE")
|
||
print("-" * 70)
|
||
songs_cols = {col['name']: col for col in inspector.get_columns('songs')}
|
||
|
||
if not songs_cols['title']['nullable']:
|
||
print(" ✅ songs.title is NOT NULL")
|
||
else:
|
||
issues.append("❌ songs.title should be NOT NULL")
|
||
fixes.append("UPDATE songs SET title = 'Untitled' WHERE title IS NULL;")
|
||
fixes.append("ALTER TABLE songs ALTER COLUMN title SET NOT NULL;")
|
||
|
||
# Check created_at/updated_at should use INTEGER (timestamps)
|
||
if str(songs_cols['created_at']['type']) == 'BIGINT':
|
||
print(" ✅ songs.created_at is BIGINT (timestamp)")
|
||
else:
|
||
issues.append(f"❌ songs.created_at type is {songs_cols['created_at']['type']}, should be BIGINT")
|
||
|
||
# Check Plans table
|
||
print("\n📋 CHECKING PLANS TABLE")
|
||
print("-" * 70)
|
||
plans_cols = {col['name']: col for col in inspector.get_columns('plans')}
|
||
|
||
if not plans_cols['date']['nullable']:
|
||
print(" ✅ plans.date is NOT NULL")
|
||
else:
|
||
issues.append("❌ plans.date should be NOT NULL")
|
||
fixes.append("UPDATE plans SET date = TO_CHAR(CURRENT_DATE, 'YYYY-MM-DD') WHERE date IS NULL;")
|
||
fixes.append("ALTER TABLE plans ALTER COLUMN date SET NOT NULL;")
|
||
|
||
# Check plan_songs table - id can be VARCHAR(255) for UUIDs or INTEGER for autoincrement
|
||
print("\n📋 CHECKING PLAN_SONGS TABLE")
|
||
print("-" * 70)
|
||
plan_songs_cols = {col['name']: col for col in inspector.get_columns('plan_songs')}
|
||
|
||
plan_songs_id_type = str(plan_songs_cols['id']['type'])
|
||
if 'VARCHAR' in plan_songs_id_type:
|
||
print(f" ✅ plan_songs.id is VARCHAR (using UUIDs)")
|
||
elif 'INTEGER' in plan_songs_id_type or 'SERIAL' in plan_songs_id_type:
|
||
print(f" ✅ plan_songs.id is INTEGER (autoincrement)")
|
||
else:
|
||
issues.append(f"❌ plan_songs.id has unexpected type: {plan_songs_id_type}")
|
||
print(f" ❌ plan_songs.id type: {plan_songs_id_type} (unexpected)")
|
||
|
||
# Check Indexes
|
||
print("\n📊 CHECKING INDEXES")
|
||
print("-" * 70)
|
||
|
||
required_indexes = {
|
||
'songs': ['idx_song_title', 'idx_song_artist', 'idx_song_band', 'idx_song_singer'],
|
||
'profiles': ['idx_profile_name'],
|
||
'plans': ['idx_plan_date', 'idx_plan_profile'],
|
||
'plan_songs': ['idx_plan_songs_order'],
|
||
'profile_songs': ['idx_profile_songs_profile'],
|
||
'profile_song_keys': ['idx_profile_song_keys']
|
||
}
|
||
|
||
for table, expected_indexes in required_indexes.items():
|
||
existing = [idx['name'] for idx in inspector.get_indexes(table)]
|
||
for idx_name in expected_indexes:
|
||
if idx_name in existing:
|
||
print(f" ✅ {table}.{idx_name}")
|
||
else:
|
||
issues.append(f"❌ Missing index: {table}.{idx_name}")
|
||
if idx_name == 'idx_song_title':
|
||
fixes.append(f"CREATE INDEX IF NOT EXISTS {idx_name} ON {table}(title);")
|
||
elif idx_name == 'idx_song_artist':
|
||
fixes.append(f"CREATE INDEX IF NOT EXISTS {idx_name} ON {table}(artist);")
|
||
elif idx_name == 'idx_song_band':
|
||
fixes.append(f"CREATE INDEX IF NOT EXISTS {idx_name} ON {table}(band);")
|
||
elif idx_name == 'idx_song_singer':
|
||
fixes.append(f"CREATE INDEX IF NOT EXISTS {idx_name} ON {table}(singer);")
|
||
elif idx_name == 'idx_profile_name':
|
||
fixes.append(f"CREATE INDEX IF NOT EXISTS {idx_name} ON {table}(name);")
|
||
elif idx_name == 'idx_plan_date':
|
||
fixes.append(f"CREATE INDEX IF NOT EXISTS {idx_name} ON {table}(date);")
|
||
elif idx_name == 'idx_plan_profile':
|
||
fixes.append(f"CREATE INDEX IF NOT EXISTS {idx_name} ON {table}(profile_id);")
|
||
elif idx_name == 'idx_plan_songs_order':
|
||
fixes.append(f"CREATE INDEX IF NOT EXISTS {idx_name} ON {table}(plan_id, order_index);")
|
||
elif idx_name == 'idx_profile_songs_profile':
|
||
fixes.append(f"CREATE INDEX IF NOT EXISTS {idx_name} ON {table}(profile_id);")
|
||
elif idx_name == 'idx_profile_song_keys':
|
||
fixes.append(f"CREATE INDEX IF NOT EXISTS {idx_name} ON {table}(profile_id, song_id);")
|
||
|
||
# Check Foreign Keys CASCADE
|
||
print("\n🔗 CHECKING FOREIGN KEY CONSTRAINTS")
|
||
print("-" * 70)
|
||
|
||
fk_tables = ['plan_songs', 'profile_songs', 'profile_song_keys']
|
||
for table in fk_tables:
|
||
fks = inspector.get_foreign_keys(table)
|
||
for fk in fks:
|
||
cascade = fk['options'].get('ondelete', 'NO ACTION')
|
||
if cascade == 'CASCADE':
|
||
print(f" ✅ {table}.{fk['name']} → CASCADE")
|
||
else:
|
||
issues.append(f"❌ {table}.{fk['name']} should CASCADE on delete (currently: {cascade})")
|
||
|
||
# Check plans.profile_id should SET NULL
|
||
plans_fks = inspector.get_foreign_keys('plans')
|
||
for fk in plans_fks:
|
||
if 'profile_id' in fk['constrained_columns']:
|
||
cascade = fk['options'].get('ondelete', 'NO ACTION')
|
||
if cascade == 'SET NULL':
|
||
print(f" ✅ plans.profile_id → SET NULL")
|
||
else:
|
||
issues.append(f"❌ plans.profile_id should SET NULL on delete (currently: {cascade})")
|
||
|
||
# Summary
|
||
print("\n" + "=" * 70)
|
||
print("ANALYSIS SUMMARY")
|
||
print("=" * 70)
|
||
|
||
if issues:
|
||
print(f"\n⚠️ Found {len(issues)} issues:\n")
|
||
for issue in issues:
|
||
print(f" {issue}")
|
||
else:
|
||
print("\n✅ No issues found! Database schema is correct.")
|
||
|
||
if fixes:
|
||
print(f"\n🔧 Suggested fixes ({len(fixes)} SQL statements):\n")
|
||
for i, fix in enumerate(fixes, 1):
|
||
print(f"{i}. {fix}")
|
||
|
||
return issues, fixes
|
||
|
||
def apply_fixes(fixes):
|
||
"""Apply database fixes"""
|
||
if not fixes:
|
||
print("\n✅ No fixes needed!")
|
||
return True
|
||
|
||
print("\n" + "=" * 70)
|
||
print("APPLYING FIXES")
|
||
print("=" * 70)
|
||
|
||
try:
|
||
with engine.begin() as conn:
|
||
for i, fix in enumerate(fixes, 1):
|
||
if fix.startswith('--'):
|
||
print(f"\n{fix}")
|
||
continue
|
||
|
||
print(f"\n{i}. Executing: {fix[:80]}...")
|
||
try:
|
||
conn.execute(text(fix))
|
||
print(" ✅ Success")
|
||
except Exception as e:
|
||
print(f" ⚠️ Warning: {str(e)}")
|
||
# Continue with other fixes
|
||
|
||
print("\n✅ All fixes applied successfully!")
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f"\n❌ Error applying fixes: {str(e)}")
|
||
return False
|
||
|
||
def verify_backend_alignment():
|
||
"""Verify backend code aligns with database schema"""
|
||
print("\n" + "=" * 70)
|
||
print("BACKEND ALIGNMENT VERIFICATION")
|
||
print("=" * 70)
|
||
|
||
inspector = inspect(engine)
|
||
|
||
# Check Profile model
|
||
print("\n📋 Profile Model vs Database")
|
||
print("-" * 70)
|
||
profile_db_cols = set(col['name'] for col in inspector.get_columns('profiles'))
|
||
profile_model_attrs = {'id', 'first_name', 'last_name', 'name', 'default_key', 'email', 'contact_number', 'notes'}
|
||
|
||
missing_in_db = profile_model_attrs - profile_db_cols
|
||
extra_in_db = profile_db_cols - profile_model_attrs - {'metadata', 'registry'}
|
||
|
||
if missing_in_db:
|
||
print(f" ⚠️ Model attributes not in DB: {missing_in_db}")
|
||
if extra_in_db:
|
||
print(f" ℹ️ DB columns not in model: {extra_in_db}")
|
||
if not missing_in_db and not extra_in_db:
|
||
print(" ✅ Profile model aligned with database")
|
||
|
||
# Check Song model
|
||
print("\n📋 Song Model vs Database")
|
||
print("-" * 70)
|
||
song_db_cols = set(col['name'] for col in inspector.get_columns('songs'))
|
||
song_model_attrs = {'id', 'title', 'artist', 'band', 'singer', 'lyrics', 'chords', 'memo', 'created_at', 'updated_at'}
|
||
|
||
missing_in_db = song_model_attrs - song_db_cols
|
||
extra_in_db = song_db_cols - song_model_attrs - {'metadata', 'registry'}
|
||
|
||
if missing_in_db:
|
||
print(f" ⚠️ Model attributes not in DB: {missing_in_db}")
|
||
if extra_in_db:
|
||
print(f" ℹ️ DB columns not in model: {extra_in_db}")
|
||
if not missing_in_db and not extra_in_db:
|
||
print(" ✅ Song model aligned with database")
|
||
|
||
print("\n" + "=" * 70)
|
||
|
||
if __name__ == "__main__":
|
||
print("\n🔍 Starting comprehensive database analysis...\n")
|
||
|
||
# Step 1: Analyze
|
||
issues, fixes = analyze_schema()
|
||
|
||
# Step 2: Apply fixes
|
||
if fixes and '--apply' in sys.argv:
|
||
print("\n⚠️ --apply flag detected, applying fixes...")
|
||
if apply_fixes(fixes):
|
||
print("\n✅ Fixes applied successfully!")
|
||
# Re-analyze to verify
|
||
print("\n🔍 Re-analyzing database...")
|
||
issues, _ = analyze_schema()
|
||
else:
|
||
print("\n❌ Some fixes failed. Please review manually.")
|
||
sys.exit(1)
|
||
elif fixes:
|
||
print("\n💡 To apply these fixes, run:")
|
||
print(f" python3 {sys.argv[0]} --apply")
|
||
|
||
# Step 3: Verify alignment
|
||
verify_backend_alignment()
|
||
|
||
# Final status
|
||
print("\n" + "=" * 70)
|
||
if issues:
|
||
print(f"⚠️ {len(issues)} issues found. Run with --apply to fix.")
|
||
sys.exit(1)
|
||
else:
|
||
print("✅ Database schema is correct and aligned!")
|
||
sys.exit(0)
|