Migration VCS conflict detection and --merge for makemigrations

import sys

import os

import operator

from optparse import make_option

from django.core.management.base import BaseCommand

from django.core.management.base import BaseCommand, CommandError

from django.core.exceptions import ImproperlyConfigured

from django.db import connections, DEFAULT_DB_ALIAS

from django.db import connections, DEFAULT_DB_ALIAS, migrations

from django.db.migrations.loader import MigrationLoader

from django.db.migrations.autodetector import MigrationAutodetector, InteractiveMigrationQuestioner

from django.db.migrations.autodetector import MigrationAutodetector

from django.db.migrations.questioner import MigrationQuestioner, InteractiveMigrationQuestioner

from django.db.migrations.state import ProjectState

from django.db.migrations.writer import MigrationWriter

from django.db.models.loading import cache

from django.utils.six.moves import reduce

class Command(BaseCommand):

    option_list = BaseCommand.option_list + (

        make_option('--dry-run', action='store_true', dest='dry_run', default=False,

            help="Just show what migrations would be made; don't actually write them."),

        make_option('--merge', action='store_true', dest='merge', default=False,

            help="Enable fixing of migration conflicts."),

    )

    help = "Creates new migration(s) for apps."

        self.verbosity = int(options.get('verbosity'))

        self.interactive = options.get('interactive')

        self.dry_run = options.get('dry_run', False)

        self.merge = options.get('merge', False)

        # Make sure the app they asked for exists

        app_labels = set(app_labels)

        # (makemigrations doesn't look at the database state).

        loader = MigrationLoader(connections[DEFAULT_DB_ALIAS])

        # Before anything else, see if there's conflicting apps and drop out

        # hard if there are any and they don't want to merge

        conflicts = loader.detect_conflicts()

        if conflicts and not self.merge:

            name_str = "; ".join(

                "%s in %s" % (", ".join(names), app)

                for app, names in conflicts.items()

            )

            raise CommandError("Conflicting migrations detected (%s).\nTo fix them run 'python manage.py makemigrations --merge'" % name_str)

        # If they want to merge and there's nothing to merge, then politely exit

        if self.merge and not conflicts:

            self.stdout.write("No conflicts detected to merge.")

            return

        # If they want to merge and there is something to merge, then

        # divert into the merge code

        if self.merge and conflicts:

            return self.handle_merge(loader, conflicts)

        # Detect changes

        autodetector = MigrationAutodetector(

            loader.graph.project_state(),

                    migration_string = writer.as_string()

                    with open(writer.path, "wb") as fh:

                        fh.write(migration_string)

    def handle_merge(self, loader, conflicts):

        """

        Handles merging together conflicted migrations interactively,

        if it's safe; otherwise, advises on how to fix it.

        """

        if self.interactive:

            questioner = InteractiveMigrationQuestioner()

        else:

            questioner = MigrationQuestioner()

        for app_label, migration_names in conflicts.items():

            # Grab out the migrations in question, and work out their

            # common ancestor.

            merge_migrations = []

            for migration_name in migration_names:

                migration = loader.get_migration(app_label, migration_name)

                migration.ancestry = loader.graph.forwards_plan((app_label, migration_name))

                merge_migrations.append(migration)

            common_ancestor = None

            for level in zip(*[m.ancestry for m in merge_migrations]):

                if reduce(operator.eq, level):

                    common_ancestor = level[0]

                else:

                    break

            if common_ancestor is None:

                raise ValueError("Could not find common ancestor of %s" % migration_names)

            # Now work out the operations along each divergent branch

            for migration in merge_migrations:

                migration.branch = migration.ancestry[

                    (migration.ancestry.index(common_ancestor) + 1):

                ]

                migration.merged_operations = []

                for node_app, node_name in migration.branch:

                    migration.merged_operations.extend(

                        loader.get_migration(node_app, node_name).operations

                    )

            # In future, this could use some of the Optimizer code

            # (can_optimize_through) to automatically see if they're

            # mergeable. For now, we always just prompt the user.

            if self.verbosity > 0:

                self.stdout.write(self.style.MIGRATE_HEADING("Merging %s" % app_label))

                for migration in merge_migrations:

                    self.stdout.write(self.style.MIGRATE_LABEL("  Branch %s" % migration.name))

                    for operation in migration.merged_operations:

                        self.stdout.write("    - %s\n" % operation.describe())

            if questioner.ask_merge(app_label):

                # If they still want to merge it, then write out an empty

                # file depending on the migrations needing merging.

                numbers = [

                    MigrationAutodetector.parse_number(migration.name)

                    for migration in merge_migrations

                ]

                try:

                    biggest_number = max([x for x in numbers if x is not None])

                except ValueError:

                    biggest_number = 1

                subclass = type("Migration", (migrations.Migration, ), {

                    "dependencies": [(app_label, migration.name) for migration in merge_migrations],

                })

                new_migration = subclass("%04i_merge" % (biggest_number + 1), app_label)

                writer = MigrationWriter(new_migration)

                with open(writer.path, "wb") as fh:

                    fh.write(writer.as_string())

                if self.verbosity > 0:

                    self.stdout.write("\nCreated new merge migration %s" % writer.path)

        # Work out which apps have migrations and which do not

        executor = MigrationExecutor(connection, self.migration_progress_callback)

        # Before anything else, see if there's conflicting apps and drop out

        # hard if there are any

        conflicts = executor.loader.detect_conflicts()

        if conflicts:

            name_str = "; ".join(

                "%s in %s" % (", ".join(names), app)

                for app, names in conflicts.items()

            )

            raise CommandError("Conflicting migrations detected (%s).\nTo fix them run 'python manage.py makemigrations --merge'" % name_str)

        # If they supplied command line arguments, work out what they mean.

        run_syncdb = False

        target_app_labels_only = True

import importlib

import os

import re

import sys

from django.db.migrations import operations

from django.db.migrations.migration import Migration

from django.db.models.loading import cache

from django.utils import datetime_safe

from django.utils.six.moves import input

from django.db.migrations.questioner import MigrationQuestioner

class MigrationAutodetector(object):

        if re.match(r"^\d+_", name):

            return int(name.split("_")[0])

        return None

class MigrationQuestioner(object):

    """

    Gives the autodetector responses to questions it might have.

    This base class has a built-in noninteractive mode, but the

    interactive subclass is what the command-line arguments will use.

    """

    def __init__(self, defaults=None):

        self.defaults = defaults or {}

    def ask_initial(self, app_label):

        "Should we create an initial migration for the app?"

        return self.defaults.get("ask_initial", False)

    def ask_not_null_addition(self, field_name, model_name):

        "Adding a NOT NULL field to a model"

        # None means quit

        return None

    def ask_rename(self, model_name, old_name, new_name, field_instance):

        "Was this field really renamed?"

        return self.defaults.get("ask_rename", False)

class InteractiveMigrationQuestioner(MigrationQuestioner):

    def __init__(self, specified_apps=set()):

        self.specified_apps = specified_apps

    def _boolean_input(self, question, default=None):

        result = input("%s " % question)

        if not result and default is not None:

            return default

        while len(result) < 1 or result[0].lower() not in "yn":

            result = input("Please answer yes or no: ")

        return result[0].lower() == "y"

    def _choice_input(self, question, choices):

        print(question)

        for i, choice in enumerate(choices):

            print(" %s) %s" % (i + 1, choice))

        result = input("Select an option: ")

        while True:

            try:

                value = int(result)

                if 0 < value <= len(choices):

                    return value

            except ValueError:

                pass

            result = input("Please select a valid option: ")

    def ask_initial(self, app_label):

        "Should we create an initial migration for the app?"

        # If it was specified on the command line, definitely true

        if app_label in self.specified_apps:

            return True

        # Otherwise, we look to see if it has a migrations module

        # without any Python files in it, apart from __init__.py.

        # Apps from the new app template will have these; the python

        # file check will ensure we skip South ones.

        models_module = cache.get_app(app_label)

        migrations_import_path = "%s.migrations" % models_module.__package__

        try:

            migrations_module = importlib.import_module(migrations_import_path)

        except ImportError:

            return False

        else:

            filenames = os.listdir(os.path.dirname(migrations_module.__file__))

            return not any(x.endswith(".py") for x in filenames if x != "__init__.py")

    def ask_not_null_addition(self, field_name, model_name):

        "Adding a NOT NULL field to a model"

        choice = self._choice_input(

            "You are trying to add a non-nullable field '%s' to %s without a default;\n" % (field_name, model_name) +

            "we can't do that (the database needs something to populate existing rows).\n" +

            "Please select a fix:",

            [

                "Provide a one-off default now (will be set on all existing rows)",

                "Quit, and let me add a default in models.py",

            ]

        )

        if choice == 2:

            sys.exit(3)

        else:

            print("Please enter the default value now, as valid Python")

            print("The datetime module is available, so you can do e.g. datetime.date.today()")

            while True:

                code = input(">>> ")

                if not code:

                    print("Please enter some code, or 'exit' (with no quotes) to exit.")

                elif code == "exit":

                    sys.exit(1)

                else:

                    try:

                        return eval(code, {}, {"datetime": datetime_safe})

                    except (SyntaxError, NameError) as e:

                        print("Invalid input: %s" % e)

                    else:

                        break

    def ask_rename(self, model_name, old_name, new_name, field_instance):

        "Was this field really renamed?"

        return self._boolean_input("Did you rename %s.%s to %s.%s (a %s)? [y/N]" % (model_name, old_name, model_name, new_name, field_instance.__class__.__name__), False)

            for parent in migration.dependencies:

                self.graph.add_dependency(key, parent)

    def detect_conflicts(self):

        """

        Looks through the loaded graph and detects any conflicts - apps

        with more than one leaf migration. Returns a dict of the app labels

        that conflict with the migration names that conflict.

        """

        seen_apps = {}

        conflicting_apps = set()

        for app_label, migration_name in self.graph.leaf_nodes():

            if app_label in seen_apps:

                conflicting_apps.add(app_label)

            seen_apps.setdefault(app_label, set()).add(migration_name)

        return dict((app_label, seen_apps[app_label]) for app_label in conflicting_apps)

class BadMigrationError(Exception):

    """

import importlib

import os

import sys

from django.db.models.loading import cache

from django.utils import datetime_safe

from django.utils.six.moves import input

from django.core.exceptions import ImproperlyConfigured

class MigrationQuestioner(object):

    """

    Gives the autodetector responses to questions it might have.

    This base class has a built-in noninteractive mode, but the

    interactive subclass is what the command-line arguments will use.

    """

    def __init__(self, defaults=None, specified_apps=None):

        self.defaults = defaults or {}

        self.specified_apps = specified_apps or set()

    def ask_initial(self, app_label):

        "Should we create an initial migration for the app?"

        # If it was specified on the command line, definitely true

        if app_label in self.specified_apps:

            return True

        # Otherwise, we look to see if it has a migrations module

        # without any Python files in it, apart from __init__.py.

        # Apps from the new app template will have these; the python

        # file check will ensure we skip South ones.

        try:

            models_module = cache.get_app(app_label)

        except ImproperlyConfigured:  # It's a fake app

            return self.defaults.get("ask_initial", False)

        migrations_import_path = "%s.migrations" % models_module.__package__

        try:

            migrations_module = importlib.import_module(migrations_import_path)

        except ImportError:

            return self.defaults.get("ask_initial", False)

        else:

            filenames = os.listdir(os.path.dirname(migrations_module.__file__))

            return not any(x.endswith(".py") for x in filenames if x != "__init__.py")

    def ask_not_null_addition(self, field_name, model_name):

        "Adding a NOT NULL field to a model"

        # None means quit

        return None

    def ask_rename(self, model_name, old_name, new_name, field_instance):

        "Was this field really renamed?"

        return self.defaults.get("ask_rename", False)

    def ask_merge(self, app_label):

        "Do you really want to merge these migrations?"

        return self.defaults.get("ask_merge", False)

class InteractiveMigrationQuestioner(MigrationQuestioner):

    def _boolean_input(self, question, default=None):

        result = input("%s " % question)

        if not result and default is not None:

            return default

        while len(result) < 1 or result[0].lower() not in "yn":

            result = input("Please answer yes or no: ")

        return result[0].lower() == "y"

    def _choice_input(self, question, choices):

        print(question)

        for i, choice in enumerate(choices):

            print(" %s) %s" % (i + 1, choice))

        result = input("Select an option: ")

        while True:

            try:

                value = int(result)

                if 0 < value <= len(choices):

                    return value

            except ValueError:

                pass

            result = input("Please select a valid option: ")

    def ask_not_null_addition(self, field_name, model_name):

        "Adding a NOT NULL field to a model"

        choice = self._choice_input(

            "You are trying to add a non-nullable field '%s' to %s without a default;\n" % (field_name, model_name) +

            "we can't do that (the database needs something to populate existing rows).\n" +

            "Please select a fix:",

            [

                "Provide a one-off default now (will be set on all existing rows)",

                "Quit, and let me add a default in models.py",

            ]

        )

        if choice == 2:

            sys.exit(3)

        else:

            print("Please enter the default value now, as valid Python")

            print("The datetime module is available, so you can do e.g. datetime.date.today()")

            while True:

                code = input(">>> ")

                if not code:

                    print("Please enter some code, or 'exit' (with no quotes) to exit.")

                elif code == "exit":

                    sys.exit(1)

                else:

                    try:

                        return eval(code, {}, {"datetime": datetime_safe})

                    except (SyntaxError, NameError) as e:

                        print("Invalid input: %s" % e)

                    else:

                        break

    def ask_rename(self, model_name, old_name, new_name, field_instance):

        "Was this field really renamed?"

        return self._boolean_input("Did you rename %s.%s to %s.%s (a %s)? [y/N]" % (model_name, old_name, model_name, new_name, field_instance.__class__.__name__), False)

    def ask_merge(self, app_label):

        return self._boolean_input(

            "\nMerging will only work if the operations printed above do not conflict\n" +

            "with each other (working on different fields or models)\n" +

            "Do you want to merge these migration branches? [y/N]",

            False,

        )