416 lines
18 KiB
Python
Executable File
416 lines
18 KiB
Python
Executable File
#!/usr/bin/env python
|
|
"""
|
|
Updated Migration Final Simple Script
|
|
====================================
|
|
|
|
This script provides a comprehensive workflow for resetting and rebuilding
|
|
Django migrations for the rogaining_srv project using a simplified approach.
|
|
|
|
Usage:
|
|
python migration_simple_reset.py [options]
|
|
|
|
Options:
|
|
--backup-only : Only create backup of existing migrations
|
|
--reset-only : Only reset migrations (requires backup to exist)
|
|
--apply-only : Only apply migrations (requires simple migration to exist)
|
|
--full : Run complete workflow (default)
|
|
|
|
Features:
|
|
- Creates timestamped backup of existing migrations
|
|
- Clears migration history from database
|
|
- Creates simplified initial migration with only managed models
|
|
- Applies migrations with proper error handling
|
|
- Provides detailed logging and status reporting
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import shutil
|
|
import datetime
|
|
import subprocess
|
|
import json
|
|
from pathlib import Path
|
|
|
|
# Add Django project to path
|
|
project_root = Path(__file__).parent
|
|
sys.path.insert(0, str(project_root))
|
|
|
|
# Setup Django environment
|
|
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'config.settings')
|
|
|
|
import django
|
|
django.setup()
|
|
|
|
from django.core.management import execute_from_command_line
|
|
from django.db import connection, transaction
|
|
from django.core.management.base import CommandError
|
|
|
|
class MigrationManager:
|
|
def __init__(self):
|
|
self.project_root = Path(__file__).parent
|
|
self.migrations_dir = self.project_root / 'rog' / 'migrations'
|
|
self.backup_timestamp = datetime.datetime.now().strftime('%Y%m%d_%H%M%S')
|
|
self.backup_dir = self.project_root / f'rog/migrations_backup_{self.backup_timestamp}'
|
|
|
|
def log(self, message, level='INFO'):
|
|
"""Log message with timestamp and level"""
|
|
timestamp = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
|
|
print(f"[{timestamp}] {level}: {message}")
|
|
|
|
def run_command(self, command, check=True):
|
|
"""Execute shell command with logging"""
|
|
self.log(f"Executing: {' '.join(command)}")
|
|
try:
|
|
result = subprocess.run(command, capture_output=True, text=True, check=check)
|
|
if result.stdout:
|
|
self.log(f"STDOUT: {result.stdout.strip()}")
|
|
if result.stderr:
|
|
self.log(f"STDERR: {result.stderr.strip()}", 'WARNING')
|
|
return result
|
|
except subprocess.CalledProcessError as e:
|
|
self.log(f"Command failed with exit code {e.returncode}", 'ERROR')
|
|
self.log(f"STDOUT: {e.stdout}", 'ERROR')
|
|
self.log(f"STDERR: {e.stderr}", 'ERROR')
|
|
raise
|
|
|
|
def backup_migrations(self):
|
|
"""Create backup of existing migrations"""
|
|
self.log("Creating backup of existing migrations...")
|
|
|
|
if not self.migrations_dir.exists():
|
|
self.log("No migrations directory found, creating empty one")
|
|
self.migrations_dir.mkdir(parents=True, exist_ok=True)
|
|
(self.migrations_dir / '__init__.py').touch()
|
|
return
|
|
|
|
# Create backup directory
|
|
self.backup_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
# Copy all files
|
|
for item in self.migrations_dir.iterdir():
|
|
if item.is_file():
|
|
shutil.copy2(item, self.backup_dir)
|
|
self.log(f"Backed up: {item.name}")
|
|
|
|
self.log(f"Backup completed: {self.backup_dir}")
|
|
|
|
def clear_migration_history(self):
|
|
"""Clear migration history from database"""
|
|
self.log("Clearing migration history from database...")
|
|
|
|
try:
|
|
with connection.cursor() as cursor:
|
|
# Check if django_migrations table exists
|
|
cursor.execute("""
|
|
SELECT COUNT(*) FROM information_schema.tables
|
|
WHERE table_name = 'django_migrations' AND table_schema = 'public'
|
|
""")
|
|
if cursor.fetchone()[0] == 0:
|
|
self.log("django_migrations table does not exist, skipping clear")
|
|
return
|
|
|
|
# Count existing migrations
|
|
cursor.execute("SELECT COUNT(*) FROM django_migrations WHERE app = 'rog'")
|
|
count = cursor.fetchone()[0]
|
|
self.log(f"Found {count} existing rog migrations")
|
|
|
|
# Delete rog migrations
|
|
cursor.execute("DELETE FROM django_migrations WHERE app = 'rog'")
|
|
self.log(f"Deleted {count} migration records")
|
|
|
|
except Exception as e:
|
|
self.log(f"Error clearing migration history: {e}", 'ERROR')
|
|
raise
|
|
|
|
def remove_migration_files(self):
|
|
"""Remove existing migration files except __init__.py"""
|
|
self.log("Removing existing migration files...")
|
|
|
|
if not self.migrations_dir.exists():
|
|
return
|
|
|
|
removed_count = 0
|
|
for item in self.migrations_dir.iterdir():
|
|
if item.is_file() and item.name != '__init__.py':
|
|
item.unlink()
|
|
self.log(f"Removed: {item.name}")
|
|
removed_count += 1
|
|
|
|
self.log(f"Removed {removed_count} migration files")
|
|
|
|
def create_simple_migration(self):
|
|
"""Create simplified initial migration"""
|
|
self.log("Creating simplified initial migration...")
|
|
|
|
# Create simple migration content
|
|
migration_content = '''# Generated by migration_simple_reset.py
|
|
from django.contrib.gis.db import models
|
|
from django.contrib.auth.models import AbstractUser
|
|
from django.db import migrations
|
|
import django.contrib.gis.db.models.fields
|
|
import django.core.validators
|
|
|
|
|
|
class Migration(migrations.Migration):
|
|
|
|
initial = True
|
|
|
|
dependencies = [
|
|
('auth', '0012_alter_user_first_name_max_length'),
|
|
]
|
|
|
|
operations = [
|
|
migrations.CreateModel(
|
|
name='CustomUser',
|
|
fields=[
|
|
('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
|
|
('password', models.CharField(max_length=128, verbose_name='password')),
|
|
('last_login', models.DateTimeField(blank=True, null=True, verbose_name='last login')),
|
|
('is_superuser', models.BooleanField(default=False, help_text='Designates that this user has all permissions without explicitly assigning them.', verbose_name='superuser status')),
|
|
('username', models.CharField(error_messages={'unique': 'A user with that username already exists.'}, help_text='Required. 150 characters or fewer. Letters, digits and @/./+/-/_ only.', max_length=150, unique=True, validators=[django.contrib.auth.validators.UnicodeUsernameValidator()], verbose_name='username')),
|
|
('first_name', models.CharField(blank=True, max_length=150, verbose_name='first name')),
|
|
('last_name', models.CharField(blank=True, max_length=150, verbose_name='last name')),
|
|
('email', models.EmailField(blank=True, max_length=254, verbose_name='email address')),
|
|
('is_staff', models.BooleanField(default=False, help_text='Designates whether the user can log into this admin site.', verbose_name='staff status')),
|
|
('is_active', models.BooleanField(default=True, help_text='Designates whether this user should be treated as active. Unselect this instead of deleting accounts.', verbose_name='active')),
|
|
('date_joined', models.DateTimeField(auto_now_add=True, verbose_name='date joined')),
|
|
('groups', models.ManyToManyField(blank=True, help_text='The groups this user belongs to. A user will get all permissions granted to each of their groups.', related_name='user_set', related_query_name='user', to='auth.Group', verbose_name='groups')),
|
|
('user_permissions', models.ManyToManyField(blank=True, help_text='Specific permissions for this user.', related_name='user_set', related_query_name='user', to='auth.Permission', verbose_name='user permissions')),
|
|
],
|
|
options={
|
|
'verbose_name': 'user',
|
|
'verbose_name_plural': 'users',
|
|
'abstract': False,
|
|
},
|
|
managers=[
|
|
('objects', django.contrib.auth.models.UserManager()),
|
|
],
|
|
),
|
|
migrations.CreateModel(
|
|
name='Category',
|
|
fields=[
|
|
('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
|
|
('name', models.CharField(max_length=50)),
|
|
('description', models.TextField(blank=True, null=True)),
|
|
('created_at', models.DateTimeField(auto_now_add=True)),
|
|
('updated_at', models.DateTimeField(auto_now=True)),
|
|
],
|
|
options={
|
|
'verbose_name_plural': 'categories',
|
|
},
|
|
),
|
|
migrations.CreateModel(
|
|
name='NewEvent',
|
|
fields=[
|
|
('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
|
|
('name', models.CharField(max_length=200)),
|
|
('description', models.TextField(blank=True, null=True)),
|
|
('event_date', models.DateField()),
|
|
('start_time', models.TimeField()),
|
|
('end_time', models.TimeField()),
|
|
('event_boundary', django.contrib.gis.db.models.fields.PolygonField(blank=True, null=True, srid=4326)),
|
|
('max_participants', models.PositiveIntegerField(default=100)),
|
|
('is_active', models.BooleanField(default=True)),
|
|
('created_at', models.DateTimeField(auto_now_add=True)),
|
|
('updated_at', models.DateTimeField(auto_now=True)),
|
|
('category', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, to='rog.category')),
|
|
],
|
|
),
|
|
migrations.CreateModel(
|
|
name='Team',
|
|
fields=[
|
|
('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
|
|
('name', models.CharField(max_length=200)),
|
|
('max_members', models.PositiveIntegerField(default=5)),
|
|
('is_active', models.BooleanField(default=True)),
|
|
('created_at', models.DateTimeField(auto_now_add=True)),
|
|
('updated_at', models.DateTimeField(auto_now=True)),
|
|
('event', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='teams', to='rog.newevent')),
|
|
],
|
|
),
|
|
migrations.CreateModel(
|
|
name='Location',
|
|
fields=[
|
|
('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
|
|
('name', models.CharField(max_length=200)),
|
|
('description', models.TextField(blank=True, null=True)),
|
|
('coordinate', django.contrib.gis.db.models.fields.PointField(srid=4326)),
|
|
('altitude', models.FloatField(blank=True, null=True)),
|
|
('location_type', models.CharField(choices=[('checkpoint', 'Checkpoint'), ('start', 'Start Point'), ('finish', 'Finish Point'), ('water', 'Water Station'), ('emergency', 'Emergency Point'), ('viewpoint', 'View Point'), ('other', 'Other')], default='checkpoint', max_length=20)),
|
|
('points', models.IntegerField(default=0)),
|
|
('is_active', models.BooleanField(default=True)),
|
|
('created_at', models.DateTimeField(auto_now_add=True)),
|
|
('updated_at', models.DateTimeField(auto_now=True)),
|
|
('event', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='locations', to='rog.newevent')),
|
|
],
|
|
),
|
|
migrations.CreateModel(
|
|
name='Entry',
|
|
fields=[
|
|
('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
|
|
('role', models.CharField(choices=[('leader', 'Team Leader'), ('member', 'Team Member')], default='member', max_length=10)),
|
|
('joined_at', models.DateTimeField(auto_now_add=True)),
|
|
('is_active', models.BooleanField(default=True)),
|
|
('team', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='entries', to='rog.team')),
|
|
('user', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='rog.customuser')),
|
|
],
|
|
options={
|
|
'verbose_name_plural': 'entries',
|
|
'unique_together': {('user', 'team')},
|
|
},
|
|
),
|
|
]
|
|
'''
|
|
|
|
# Write migration file
|
|
migration_file = self.migrations_dir / '0001_simple_initial.py'
|
|
with open(migration_file, 'w', encoding='utf-8') as f:
|
|
f.write(migration_content)
|
|
|
|
self.log(f"Created: {migration_file}")
|
|
|
|
def apply_migrations(self):
|
|
"""Apply migrations using Django management command"""
|
|
self.log("Applying migrations...")
|
|
|
|
try:
|
|
# Use fake-initial to treat as initial migration
|
|
result = self.run_command([
|
|
'python', 'manage.py', 'migrate', '--fake-initial'
|
|
], check=False)
|
|
|
|
if result.returncode == 0:
|
|
self.log("Migrations applied successfully")
|
|
else:
|
|
self.log("Migration application failed", 'ERROR')
|
|
raise CommandError("Migration application failed")
|
|
|
|
except Exception as e:
|
|
self.log(f"Error applying migrations: {e}", 'ERROR')
|
|
raise
|
|
|
|
def check_status(self):
|
|
"""Check current migration status"""
|
|
self.log("Checking migration status...")
|
|
|
|
try:
|
|
result = self.run_command([
|
|
'python', 'manage.py', 'showmigrations', 'rog'
|
|
], check=False)
|
|
|
|
if result.returncode == 0:
|
|
self.log("Migration status checked successfully")
|
|
else:
|
|
self.log("Failed to check migration status", 'WARNING')
|
|
|
|
except Exception as e:
|
|
self.log(f"Error checking migration status: {e}", 'WARNING')
|
|
|
|
def run_full_workflow(self):
|
|
"""Execute complete migration reset workflow"""
|
|
self.log("Starting full migration reset workflow...")
|
|
|
|
try:
|
|
# Step 1: Backup existing migrations
|
|
self.backup_migrations()
|
|
|
|
# Step 2: Clear migration history
|
|
self.clear_migration_history()
|
|
|
|
# Step 3: Remove migration files
|
|
self.remove_migration_files()
|
|
|
|
# Step 4: Create simplified migration
|
|
self.create_simple_migration()
|
|
|
|
# Step 5: Apply migrations
|
|
self.apply_migrations()
|
|
|
|
# Step 6: Check final status
|
|
self.check_status()
|
|
|
|
self.log("Full migration reset workflow completed successfully!")
|
|
self.log(f"Backup created at: {self.backup_dir}")
|
|
|
|
except Exception as e:
|
|
self.log(f"Workflow failed: {e}", 'ERROR')
|
|
self.log(f"You can restore from backup at: {self.backup_dir}", 'INFO')
|
|
raise
|
|
|
|
def run_backup_only(self):
|
|
"""Create backup only"""
|
|
self.log("Running backup-only mode...")
|
|
self.backup_migrations()
|
|
self.log("Backup completed successfully!")
|
|
|
|
def run_reset_only(self):
|
|
"""Reset migrations only (requires backup)"""
|
|
self.log("Running reset-only mode...")
|
|
self.clear_migration_history()
|
|
self.remove_migration_files()
|
|
self.create_simple_migration()
|
|
self.log("Reset completed successfully!")
|
|
|
|
def run_apply_only(self):
|
|
"""Apply migrations only"""
|
|
self.log("Running apply-only mode...")
|
|
self.apply_migrations()
|
|
self.check_status()
|
|
self.log("Apply completed successfully!")
|
|
|
|
|
|
def main():
|
|
"""Main function with command line argument handling"""
|
|
import argparse
|
|
|
|
parser = argparse.ArgumentParser(
|
|
description='Reset and rebuild Django migrations for rogaining_srv project'
|
|
)
|
|
parser.add_argument(
|
|
'--backup-only',
|
|
action='store_true',
|
|
help='Only create backup of existing migrations'
|
|
)
|
|
parser.add_argument(
|
|
'--reset-only',
|
|
action='store_true',
|
|
help='Only reset migrations (requires backup to exist)'
|
|
)
|
|
parser.add_argument(
|
|
'--apply-only',
|
|
action='store_true',
|
|
help='Only apply migrations (requires simple migration to exist)'
|
|
)
|
|
parser.add_argument(
|
|
'--full',
|
|
action='store_true',
|
|
help='Run complete workflow (default)'
|
|
)
|
|
|
|
args = parser.parse_args()
|
|
|
|
# If no specific mode is selected, default to full
|
|
if not any([args.backup_only, args.reset_only, args.apply_only, args.full]):
|
|
args.full = True
|
|
|
|
manager = MigrationManager()
|
|
|
|
try:
|
|
if args.backup_only:
|
|
manager.run_backup_only()
|
|
elif args.reset_only:
|
|
manager.run_reset_only()
|
|
elif args.apply_only:
|
|
manager.run_apply_only()
|
|
elif args.full:
|
|
manager.run_full_workflow()
|
|
|
|
except Exception as e:
|
|
print(f"\nError: {e}")
|
|
print("Please check the logs above for details.")
|
|
sys.exit(1)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|