Fix deployment issue, see DEPLOYMENT_MIGRATION_GUIDE_en.md
This commit is contained in:
415
migration_simple_reset.py
Executable file
415
migration_simple_reset.py
Executable file
@ -0,0 +1,415 @@
|
||||
#!/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()
|
||||
Reference in New Issue
Block a user