239 lines
9.8 KiB
Python
Executable File
239 lines
9.8 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
Database Migration Status Checker
|
|
Checks what migration steps have been completed and what's missing
|
|
"""
|
|
import sys
|
|
import os
|
|
from sqlalchemy import create_engine, text, inspect
|
|
from sqlalchemy.orm import sessionmaker
|
|
from dotenv import load_dotenv
|
|
|
|
# Load environment variables
|
|
load_dotenv()
|
|
|
|
# Get database URL from environment or use provided one
|
|
DATABASE_URL = os.getenv('DATABASE_URL')
|
|
|
|
if not DATABASE_URL:
|
|
print("ERROR: DATABASE_URL not found in environment")
|
|
sys.exit(1)
|
|
|
|
# Create database connection
|
|
engine = create_engine(DATABASE_URL)
|
|
Session = sessionmaker(bind=engine)
|
|
db = Session()
|
|
inspector = inspect(engine)
|
|
|
|
print("=" * 80)
|
|
print("DATABASE MIGRATION STATUS CHECKER")
|
|
print("=" * 80)
|
|
print(f"\nConnected to: {DATABASE_URL.split('@')[1] if '@' in DATABASE_URL else 'database'}")
|
|
print()
|
|
|
|
# Colors for output
|
|
class Colors:
|
|
GREEN = '\033[92m'
|
|
RED = '\033[91m'
|
|
YELLOW = '\033[93m'
|
|
BLUE = '\033[94m'
|
|
END = '\033[0m'
|
|
|
|
def check_mark(exists):
|
|
return f"{Colors.GREEN}✓{Colors.END}" if exists else f"{Colors.RED}✗{Colors.END}"
|
|
|
|
def warning(exists):
|
|
return f"{Colors.YELLOW}⚠{Colors.END}" if exists else f"{Colors.GREEN}✓{Colors.END}"
|
|
|
|
issues = []
|
|
warnings = []
|
|
|
|
# ============================================================
|
|
# Check 1: Does roles table exist?
|
|
# ============================================================
|
|
print(f"{Colors.BLUE}[1] Checking if 'roles' table exists...{Colors.END}")
|
|
roles_table_exists = 'roles' in inspector.get_table_names()
|
|
print(f" {check_mark(roles_table_exists)} roles table exists")
|
|
|
|
if not roles_table_exists:
|
|
issues.append("❌ MISSING: 'roles' table - run migration 006_add_dynamic_roles.sql")
|
|
print(f"\n{Colors.RED}ISSUE: roles table not found!{Colors.END}")
|
|
print(f" Action: Run 'psql $DATABASE_URL -f migrations/006_add_dynamic_roles.sql'")
|
|
|
|
# ============================================================
|
|
# Check 2: Does users table have role_id column?
|
|
# ============================================================
|
|
print(f"\n{Colors.BLUE}[2] Checking if 'users' table has 'role_id' column...{Colors.END}")
|
|
users_columns = [col['name'] for col in inspector.get_columns('users')]
|
|
users_has_role_id = 'role_id' in users_columns
|
|
print(f" {check_mark(users_has_role_id)} users.role_id column exists")
|
|
|
|
if not users_has_role_id:
|
|
issues.append("❌ MISSING: 'users.role_id' column - run migration 006_add_dynamic_roles.sql")
|
|
print(f"\n{Colors.RED}ISSUE: users.role_id column not found!{Colors.END}")
|
|
print(f" Action: Run 'psql $DATABASE_URL -f migrations/006_add_dynamic_roles.sql'")
|
|
|
|
# ============================================================
|
|
# Check 3: Does role_permissions table have role_id column?
|
|
# ============================================================
|
|
print(f"\n{Colors.BLUE}[3] Checking if 'role_permissions' table has 'role_id' column...{Colors.END}")
|
|
rp_columns = [col['name'] for col in inspector.get_columns('role_permissions')]
|
|
rp_has_role_id = 'role_id' in rp_columns
|
|
print(f" {check_mark(rp_has_role_id)} role_permissions.role_id column exists")
|
|
|
|
if not rp_has_role_id:
|
|
issues.append("❌ MISSING: 'role_permissions.role_id' column - run migration 006_add_dynamic_roles.sql")
|
|
print(f"\n{Colors.RED}ISSUE: role_permissions.role_id column not found!{Colors.END}")
|
|
print(f" Action: Run 'psql $DATABASE_URL -f migrations/006_add_dynamic_roles.sql'")
|
|
|
|
# ============================================================
|
|
# Check 4: Are system roles seeded?
|
|
# ============================================================
|
|
if roles_table_exists:
|
|
print(f"\n{Colors.BLUE}[4] Checking if system roles are seeded...{Colors.END}")
|
|
result = db.execute(text("SELECT COUNT(*) as count FROM roles WHERE is_system_role = true"))
|
|
system_roles_count = result.scalar()
|
|
print(f" System roles found: {system_roles_count}")
|
|
|
|
if system_roles_count == 0:
|
|
issues.append("❌ MISSING: System roles not seeded - run roles_seed.py")
|
|
print(f" {Colors.RED}✗ No system roles found!{Colors.END}")
|
|
print(f" Action: Run 'python3 roles_seed.py'")
|
|
elif system_roles_count < 5:
|
|
warnings.append(f"⚠️ WARNING: Expected 5 system roles, found {system_roles_count}")
|
|
print(f" {Colors.YELLOW}⚠{Colors.END} Expected 5 roles, found {system_roles_count}")
|
|
print(f" Action: Run 'python3 roles_seed.py' to ensure all roles exist")
|
|
else:
|
|
print(f" {Colors.GREEN}✓{Colors.END} All system roles seeded")
|
|
|
|
# Show which roles exist
|
|
result = db.execute(text("SELECT code, name FROM roles WHERE is_system_role = true ORDER BY code"))
|
|
existing_roles = result.fetchall()
|
|
if existing_roles:
|
|
print(f"\n Existing roles:")
|
|
for role in existing_roles:
|
|
print(f" - {role[0]}: {role[1]}")
|
|
|
|
# ============================================================
|
|
# Check 5: Are users migrated to dynamic roles?
|
|
# ============================================================
|
|
if users_has_role_id and roles_table_exists:
|
|
print(f"\n{Colors.BLUE}[5] Checking if users are migrated to dynamic roles...{Colors.END}")
|
|
|
|
# Count total users
|
|
result = db.execute(text("SELECT COUNT(*) FROM users"))
|
|
total_users = result.scalar()
|
|
print(f" Total users: {total_users}")
|
|
|
|
# Count users with role_id set
|
|
result = db.execute(text("SELECT COUNT(*) FROM users WHERE role_id IS NOT NULL"))
|
|
migrated_users = result.scalar()
|
|
print(f" Migrated users (with role_id): {migrated_users}")
|
|
|
|
# Count users without role_id
|
|
unmigrated_users = total_users - migrated_users
|
|
|
|
if unmigrated_users > 0:
|
|
issues.append(f"❌ INCOMPLETE: {unmigrated_users} users not migrated to dynamic roles")
|
|
print(f" {Colors.RED}✗ {unmigrated_users} users still need migration!{Colors.END}")
|
|
print(f" Action: Run 'python3 migrate_users_to_dynamic_roles.py'")
|
|
|
|
# Show sample unmigrated users
|
|
result = db.execute(text("""
|
|
SELECT email, role FROM users
|
|
WHERE role_id IS NULL
|
|
LIMIT 5
|
|
"""))
|
|
unmigrated = result.fetchall()
|
|
if unmigrated:
|
|
print(f"\n Sample unmigrated users:")
|
|
for user in unmigrated:
|
|
print(f" - {user[0]} (role: {user[1]})")
|
|
else:
|
|
print(f" {Colors.GREEN}✓{Colors.END} All users migrated to dynamic roles")
|
|
|
|
# ============================================================
|
|
# Check 6: Are role permissions migrated?
|
|
# ============================================================
|
|
if rp_has_role_id and roles_table_exists:
|
|
print(f"\n{Colors.BLUE}[6] Checking if role permissions are migrated...{Colors.END}")
|
|
|
|
# Count total role_permissions
|
|
result = db.execute(text("SELECT COUNT(*) FROM role_permissions"))
|
|
total_perms = result.scalar()
|
|
print(f" Total role_permissions: {total_perms}")
|
|
|
|
# Count permissions with role_id set
|
|
result = db.execute(text("SELECT COUNT(*) FROM role_permissions WHERE role_id IS NOT NULL"))
|
|
migrated_perms = result.scalar()
|
|
print(f" Migrated permissions (with role_id): {migrated_perms}")
|
|
|
|
unmigrated_perms = total_perms - migrated_perms
|
|
|
|
if unmigrated_perms > 0:
|
|
issues.append(f"❌ INCOMPLETE: {unmigrated_perms} permissions not migrated to dynamic roles")
|
|
print(f" {Colors.RED}✗ {unmigrated_perms} permissions still need migration!{Colors.END}")
|
|
print(f" Action: Run 'python3 migrate_role_permissions_to_dynamic_roles.py'")
|
|
else:
|
|
print(f" {Colors.GREEN}✓{Colors.END} All permissions migrated to dynamic roles")
|
|
|
|
# ============================================================
|
|
# Check 7: Verify admin account
|
|
# ============================================================
|
|
print(f"\n{Colors.BLUE}[7] Checking admin account...{Colors.END}")
|
|
result = db.execute(text("""
|
|
SELECT email, role, role_id
|
|
FROM users
|
|
WHERE email LIKE '%admin%' OR role = 'admin' OR role = 'superadmin'
|
|
LIMIT 5
|
|
"""))
|
|
admin_users = result.fetchall()
|
|
|
|
if admin_users:
|
|
print(f" Found {len(admin_users)} admin/superadmin users:")
|
|
for user in admin_users:
|
|
role_id_status = "✓" if user[2] else "✗"
|
|
print(f" {role_id_status} {user[0]} (role: {user[1]}, role_id: {user[2] or 'NULL'})")
|
|
else:
|
|
warnings.append("⚠️ WARNING: No admin users found")
|
|
print(f" {Colors.YELLOW}⚠{Colors.END} No admin users found in database")
|
|
|
|
# ============================================================
|
|
# Summary
|
|
# ============================================================
|
|
print("\n" + "=" * 80)
|
|
print("SUMMARY")
|
|
print("=" * 80)
|
|
|
|
if not issues and not warnings:
|
|
print(f"\n{Colors.GREEN}✓ All migration steps completed successfully!{Colors.END}")
|
|
print("\nNext steps:")
|
|
print(" 1. Deploy latest backend code")
|
|
print(" 2. Restart backend server")
|
|
print(" 3. Test /api/admin/users/export endpoint")
|
|
else:
|
|
if issues:
|
|
print(f"\n{Colors.RED}ISSUES FOUND ({len(issues)}):{Colors.END}")
|
|
for i, issue in enumerate(issues, 1):
|
|
print(f" {i}. {issue}")
|
|
|
|
if warnings:
|
|
print(f"\n{Colors.YELLOW}WARNINGS ({len(warnings)}):{Colors.END}")
|
|
for i, warning in enumerate(warnings, 1):
|
|
print(f" {i}. {warning}")
|
|
|
|
print(f"\n{Colors.BLUE}RECOMMENDED ACTIONS:{Colors.END}")
|
|
if not roles_table_exists or not users_has_role_id or not rp_has_role_id:
|
|
print(" 1. Run: psql $DATABASE_URL -f migrations/006_add_dynamic_roles.sql")
|
|
if roles_table_exists and system_roles_count == 0:
|
|
print(" 2. Run: python3 roles_seed.py")
|
|
if unmigrated_users > 0:
|
|
print(" 3. Run: python3 migrate_users_to_dynamic_roles.py")
|
|
if unmigrated_perms > 0:
|
|
print(" 4. Run: python3 migrate_role_permissions_to_dynamic_roles.py")
|
|
print(" 5. Deploy latest backend code and restart server")
|
|
|
|
print("\n" + "=" * 80)
|
|
|
|
db.close()
|