import sys from optparse import make_option from django.core.management.base import BaseCommand, CommandError from django.utils import six from django.db import connections, DEFAULT_DB_ALIAS, migrations from django.db.migrations.loader import AmbiguityError from django.db.migrations.executor import MigrationExecutor from django.db.migrations.writer import MigrationWriter from django.db.migrations.optimizer import MigrationOptimizer class Command(BaseCommand): option_list = BaseCommand.option_list + ( make_option('--no-optimize', action='store_true', dest='no_optimize', default=False, help='Do not try to optimize the squashed operations.'), make_option('--noinput', action='store_false', dest='interactive', default=True, help='Tells Django to NOT prompt the user for input of any kind.'), ) help = "Squashes an existing set of migrations (from first until specified) into a single new one." usage_str = "Usage: ./manage.py squashmigrations app migration_name" def handle(self, app_label=None, migration_name=None, **options): self.verbosity = int(options.get('verbosity')) self.interactive = options.get('interactive') if app_label is None or migration_name is None: self.stderr.write(self.usage_str) sys.exit(1) # Load the current graph state, check the app and migration they asked for exists executor = MigrationExecutor(connections[DEFAULT_DB_ALIAS]) if app_label not in executor.loader.migrated_apps: raise CommandError("App '%s' does not have migrations (so squashmigrations on it makes no sense)" % app_label) try: migration = executor.loader.get_migration_by_prefix(app_label, migration_name) except AmbiguityError: raise CommandError("More than one migration matches '%s' in app '%s'. Please be more specific." % (migration_name, app_label)) except KeyError: raise CommandError("Cannot find a migration matching '%s' from app '%s'." % (migration_name, app_label)) # Work out the list of predecessor migrations migrations_to_squash = [ executor.loader.get_migration(al, mn) for al, mn in executor.loader.graph.forwards_plan((migration.app_label, migration.name)) if al == migration.app_label ] # Tell them what we're doing and optionally ask if we should proceed if self.verbosity > 0 or self.interactive: self.stdout.write(self.style.MIGRATE_HEADING("Will squash the following migrations:")) for migration in migrations_to_squash: self.stdout.write(" - %s" % migration.name) if self.interactive: answer = None while not answer or answer not in "yn": answer = six.moves.input("Do you wish to proceed? [yN] ") if not answer: answer = "n" break else: answer = answer[0].lower() if answer != "y": return # Load the operations from all those migrations and concat together operations = [] for smigration in migrations_to_squash: operations.extend(smigration.operations) if self.verbosity > 0: self.stdout.write(self.style.MIGRATE_HEADING("Optimizing...")) optimizer = MigrationOptimizer() new_operations = optimizer.optimize(operations, migration.app_label) if self.verbosity > 0: if len(new_operations) == len(operations): self.stdout.write(" No optimizations possible.") else: self.stdout.write(" Optimized from %s operations to %s operations." % (len(operations), len(new_operations))) # Work out the value of replaces (any squashed ones we're re-squashing) # need to feed their replaces into ours replaces = [] for migration in migrations_to_squash: if migration.replaces: replaces.extend(migration.replaces) else: replaces.append((migration.app_label, migration.name)) # Make a new migration with those operations subclass = type("Migration", (migrations.Migration, ), { "dependencies": [], "operations": new_operations, "replaces": replaces, }) new_migration = subclass("0001_squashed_%s" % migration.name, app_label) # Write out the new migration file writer = MigrationWriter(new_migration) with open(writer.path, "wb") as fh: fh.write(writer.as_string()) if self.verbosity > 0: self.stdout.write(self.style.MIGRATE_HEADING("Created new squashed migration %s" % writer.path)) self.stdout.write(" You should commit this migration but leave the old ones in place;") self.stdout.write(" the new migration will be used for new installs. Once you are sure") self.stdout.write(" all instances of the codebase have applied the migrations you squashed,") self.stdout.write(" you can delete them.")