142 lines
4.6 KiB
Python
142 lines
4.6 KiB
Python
"""
|
|
User Role Migration Script (Phase 3)
|
|
|
|
This script migrates existing users from the legacy role enum to the new dynamic role system.
|
|
For each user, it maps their current role enum value to the corresponding role_id in the roles table.
|
|
|
|
Usage:
|
|
python migrate_users_to_dynamic_roles.py
|
|
|
|
Environment Variables:
|
|
DATABASE_URL - PostgreSQL connection string
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
from database import Base
|
|
from models import User, Role, UserRole
|
|
from dotenv import load_dotenv
|
|
|
|
# Load environment variables
|
|
load_dotenv()
|
|
|
|
# Database connection
|
|
DATABASE_URL = os.getenv("DATABASE_URL")
|
|
if not DATABASE_URL:
|
|
print("Error: DATABASE_URL environment variable not set")
|
|
sys.exit(1)
|
|
|
|
engine = create_engine(DATABASE_URL)
|
|
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
|
|
|
|
|
|
def migrate_users():
|
|
"""Migrate users from enum role to role_id"""
|
|
db = SessionLocal()
|
|
|
|
try:
|
|
print("🚀 Starting user role migration (Phase 3)...")
|
|
print("="*60)
|
|
|
|
# Step 1: Load all roles into a map
|
|
print("\n📋 Loading roles from database...")
|
|
roles = db.query(Role).all()
|
|
role_map = {role.code: role for role in roles}
|
|
|
|
print(f"✓ Loaded {len(roles)} roles:")
|
|
for role in roles:
|
|
print(f" • {role.name} ({role.code}) - ID: {role.id}")
|
|
|
|
# Step 2: Get all users
|
|
print("\n👥 Loading users...")
|
|
users = db.query(User).all()
|
|
print(f"✓ Found {len(users)} users to migrate")
|
|
|
|
# Step 3: Check if any users already have role_id set
|
|
users_with_role_id = [u for u in users if u.role_id is not None]
|
|
if users_with_role_id:
|
|
print(f"\n⚠️ Warning: {len(users_with_role_id)} users already have role_id set")
|
|
response = input("Do you want to re-migrate these users? (yes/no): ")
|
|
if response.lower() != 'yes':
|
|
print("Skipping users that already have role_id set...")
|
|
users = [u for u in users if u.role_id is None]
|
|
print(f"Will migrate {len(users)} users without role_id")
|
|
|
|
if not users:
|
|
print("\n✅ No users to migrate!")
|
|
return
|
|
|
|
# Step 4: Migrate users
|
|
print(f"\n🔄 Migrating {len(users)} users...")
|
|
|
|
migration_stats = {
|
|
UserRole.guest: 0,
|
|
UserRole.member: 0,
|
|
UserRole.admin: 0,
|
|
UserRole.superadmin: 0
|
|
}
|
|
|
|
for user in users:
|
|
# Get the enum role code (e.g., "guest", "member", "admin", "superadmin")
|
|
role_code = user.role.value
|
|
|
|
# Find the matching role in the roles table
|
|
if role_code not in role_map:
|
|
print(f" ⚠️ Warning: No matching role found for '{role_code}' (user: {user.email})")
|
|
continue
|
|
|
|
# Set the role_id
|
|
user.role_id = role_map[role_code].id
|
|
migration_stats[user.role] = migration_stats.get(user.role, 0) + 1
|
|
|
|
# Commit all changes
|
|
db.commit()
|
|
print(f"✓ Migrated {len(users)} users")
|
|
|
|
# Step 5: Display migration summary
|
|
print("\n" + "="*60)
|
|
print("📊 Migration Summary:")
|
|
print("="*60)
|
|
print("\nUsers migrated by role:")
|
|
for role_enum, count in migration_stats.items():
|
|
if count > 0:
|
|
print(f" • {role_enum.value}: {count} users")
|
|
|
|
# Step 6: Verify migration
|
|
print("\n🔍 Verifying migration...")
|
|
users_without_role_id = db.query(User).filter(User.role_id == None).count()
|
|
users_with_role_id = db.query(User).filter(User.role_id != None).count()
|
|
|
|
print(f" • Users with role_id: {users_with_role_id}")
|
|
print(f" • Users without role_id: {users_without_role_id}")
|
|
|
|
if users_without_role_id > 0:
|
|
print(f"\n⚠️ Warning: {users_without_role_id} users still don't have role_id set!")
|
|
else:
|
|
print("\n✅ All users successfully migrated!")
|
|
|
|
print("\n" + "="*60)
|
|
print("✅ User migration completed successfully!")
|
|
print("="*60)
|
|
|
|
print("\n📝 Next Steps:")
|
|
print(" 1. Migrate role_permissions table")
|
|
print(" 2. Update auth.py to use dynamic roles")
|
|
print(" 3. Update server.py role checks")
|
|
print(" 4. Verify system still works with new roles")
|
|
|
|
except Exception as e:
|
|
db.rollback()
|
|
print(f"\n❌ Error migrating users: {str(e)}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
raise
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
migrate_users()
|