if logs and prompts are diverted to stderr remove the error style make sure the app they asked for exists load the current graph state pass in none for the connection so the loader doesn t try to resolve replaced migrations from db raise an error if any migrations are applied before their dependencies nondefault databases are only checked if database routers used at least one model must be migrated to the database before anything else see if there s conflicting apps and drop out hard if there are any and they don t want to merge if applabels is specified filter out conflicting migrations for unspecified apps if they want to merge and there s nothing to merge then politely exit if they want to merge and there is something to merge then divert into the merge code set up autodetector if they want to make an empty migration make one for each app make a fake changes result we can pass to arrangeforgraph detect changes no changes tell them find last migration multiple leaf nodes have already been checked earlier in command updated migration cannot be a squash migration a dependency of another migration and cannot be already applied build new migration optimize migration update name register overridden migration take a changes dict and write them out as migration files describe the migration display a relative path if it s below the current working directory or an absolute path otherwise write the migrations file to the disk we just do this once per app alternatively makemigrations dryrun verbosity 3 will log the migrations rather than saving the file to the disk handles merging together conflicted migrations interactively if it s safe otherwise advises on how to fix it grab out the migrations in question and work out their common ancestor now work out the operations along each divergent branch in future this could use some of the optimizer code canoptimizethrough to automatically see if they re mergeable for now we always just prompt the user if they still want to merge it then write out an empty file depending on the migrations needing merging write the merge migrations file to the disk alternatively makemigrations merge dryrun verbosity 3 will log the merge migrations rather than saving the file to the disk if logs and prompts are diverted to stderr remove the error style make sure the app they asked for exists load the current graph state pass in none for the connection so the loader doesn t try to resolve replaced migrations from db raise an error if any migrations are applied before their dependencies non default databases are only checked if database routers used at least one model must be migrated to the database before anything else see if there s conflicting apps and drop out hard if there are any and they don t want to merge if app_labels is specified filter out conflicting migrations for unspecified apps if they want to merge and there s nothing to merge then politely exit if they want to merge and there is something to merge then divert into the merge code set up autodetector if they want to make an empty migration make one for each app make a fake changes result we can pass to arrange_for_graph detect changes no changes tell them find last migration multiple leaf nodes have already been checked earlier in command updated migration cannot be a squash migration a dependency of another migration and cannot be already applied build new migration optimize migration update name register overridden migration take a changes dict and write them out as migration files describe the migration display a relative path if it s below the current working directory or an absolute path otherwise write the migrations file to the disk we just do this once per app alternatively makemigrations dry run verbosity 3 will log the migrations rather than saving the file to the disk handles merging together conflicted migrations interactively if it s safe otherwise advises on how to fix it grab out the migrations in question and work out their common ancestor now work out the operations along each divergent branch 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 they still want to merge it then write out an empty file depending on the migrations needing merging write the merge migrations file to the disk alternatively makemigrations merge dry run verbosity 3 will log the merge migrations rather than saving the file to the disk
import os
import sys
import warnings
from itertools import takewhile

from django.apps import apps
from django.conf import settings
from django.core.management.base import BaseCommand, CommandError, no_translations
from django.core.management.utils import run_formatters
from django.db import DEFAULT_DB_ALIAS, OperationalError, connections, router
from django.db.migrations import Migration
from django.db.migrations.autodetector import MigrationAutodetector
from django.db.migrations.loader import MigrationLoader
from django.db.migrations.migration import SwappableTuple
from django.db.migrations.optimizer import MigrationOptimizer
from django.db.migrations.questioner import (
    InteractiveMigrationQuestioner,
    MigrationQuestioner,
    NonInteractiveMigrationQuestioner,
)
from django.db.migrations.state import ProjectState
from django.db.migrations.utils import get_migration_name_timestamp
from django.db.migrations.writer import MigrationWriter


class Command(BaseCommand):
    help = "Creates new migration(s) for apps."

    def add_arguments(self, parser):
        parser.add_argument(
            "args",
            metavar="app_label",
            nargs="*",
            help="Specify the app label(s) to create migrations for.",
        )
        parser.add_argument(
            "--dry-run",
            action="store_true",
            help="Just show what migrations would be made; don't actually write them.",
        )
        parser.add_argument(
            "--merge",
            action="store_true",
            help="Enable fixing of migration conflicts.",
        )
        parser.add_argument(
            "--empty",
            action="store_true",
            help="Create an empty migration.",
        )
        parser.add_argument(
            "--noinput",
            "--no-input",
            action="store_false",
            dest="interactive",
            help="Tells Django to NOT prompt the user for input of any kind.",
        )
        parser.add_argument(
            "-n",
            "--name",
            help="Use this name for migration file(s).",
        )
        parser.add_argument(
            "--no-header",
            action="store_false",
            dest="include_header",
            help="Do not add header comments to new migration file(s).",
        )
        parser.add_argument(
            "--check",
            action="store_true",
            dest="check_changes",
            help=(
                "Exit with a non-zero status if model changes are missing migrations "
                "and don't actually write them. Implies --dry-run."
            ),
        )
        parser.add_argument(
            "--scriptable",
            action="store_true",
            dest="scriptable",
            help=(
                "Divert log output and input prompts to stderr, writing only "
                "paths of generated migration files to stdout."
            ),
        )
        parser.add_argument(
            "--update",
            action="store_true",
            dest="update",
            help=(
                "Merge model changes into the latest migration and optimize the "
                "resulting operations."
            ),
        )

    @property
    def log_output(self):
        return self.stderr if self.scriptable else self.stdout

    def log(self, msg):
        self.log_output.write(msg)

    @no_translations
    def handle(self, *app_labels, **options):
        self.written_files = []
        self.verbosity = options["verbosity"]
        self.interactive = options["interactive"]
        self.dry_run = options["dry_run"]
        self.merge = options["merge"]
        self.empty = options["empty"]
        self.migration_name = options["name"]
        if self.migration_name and not self.migration_name.isidentifier():
            raise CommandError("The migration name must be a valid Python identifier.")
        self.include_header = options["include_header"]
        check_changes = options["check_changes"]
        if check_changes:
            self.dry_run = True
        self.scriptable = options["scriptable"]
        self.update = options["update"]
        
        if self.scriptable:
            self.stderr.style_func = None

        
        app_labels = set(app_labels)
        has_bad_labels = False
        for app_label in app_labels:
            try:
                apps.get_app_config(app_label)
            except LookupError as err:
                self.stderr.write(str(err))
                has_bad_labels = True
        if has_bad_labels:
            sys.exit(2)

        
        
        loader = MigrationLoader(None, ignore_no_migrations=True)

        
        consistency_check_labels = {config.label for config in apps.get_app_configs()}
        
        aliases_to_check = (
            connections if settings.DATABASE_ROUTERS else [DEFAULT_DB_ALIAS]
        )
        for alias in sorted(aliases_to_check):
            connection = connections[alias]
            if connection.settings_dict["ENGINE"] != "django.db.backends.dummy" and any(
                
                router.allow_migrate(
                    connection.alias, app_label, model_name=model._meta.object_name
                )
                for app_label in consistency_check_labels
                for model in apps.get_app_config(app_label).get_models()
            ):
                try:
                    loader.check_consistent_history(connection)
                except OperationalError as error:
                    warnings.warn(
                        "Got an error checking a consistent migration history "
                        "performed for database connection '%s': %s" % (alias, error),
                        RuntimeWarning,
                    )
        
        
        conflicts = loader.detect_conflicts()

        
        
        if app_labels:
            conflicts = {
                app_label: conflict
                for app_label, conflict in conflicts.items()
                if app_label in app_labels
            }

        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; multiple leaf nodes in the "
                "migration graph: (%s).\nTo fix them run "
                "'python manage.py makemigrations --merge'" % name_str
            )

        
        if self.merge and not conflicts:
            self.log("No conflicts detected to merge.")
            return

        
        
        if self.merge and conflicts:
            return self.handle_merge(loader, conflicts)

        if self.interactive:
            questioner = InteractiveMigrationQuestioner(
                specified_apps=app_labels,
                dry_run=self.dry_run,
                prompt_output=self.log_output,
            )
        else:
            questioner = NonInteractiveMigrationQuestioner(
                specified_apps=app_labels,
                dry_run=self.dry_run,
                verbosity=self.verbosity,
                log=self.log,
            )
        
        autodetector = MigrationAutodetector(
            loader.project_state(),
            ProjectState.from_apps(apps),
            questioner,
        )

        
        if self.empty:
            if not app_labels:
                raise CommandError(
                    "You must supply at least one app label when using --empty."
                )
            
            changes = {app: [Migration("custom", app)] for app in app_labels}
            changes = autodetector.arrange_for_graph(
                changes=changes,
                graph=loader.graph,
                migration_name=self.migration_name,
            )
            self.write_migration_files(changes)
            return

        
        changes = autodetector.changes(
            graph=loader.graph,
            trim_to_apps=app_labels or None,
            convert_apps=app_labels or None,
            migration_name=self.migration_name,
        )

        if not changes:
            
            if self.verbosity >= 1:
                if app_labels:
                    if len(app_labels) == 1:
                        self.log("No changes detected in app '%s'" % app_labels.pop())
                    else:
                        self.log(
                            "No changes detected in apps '%s'"
                            % ("', '".join(app_labels))
                        )
                else:
                    self.log("No changes detected")
        else:
            if self.update:
                self.write_to_last_migration_files(changes)
            else:
                self.write_migration_files(changes)
            if check_changes:
                sys.exit(1)

    def write_to_last_migration_files(self, changes):
        loader = MigrationLoader(connections[DEFAULT_DB_ALIAS])
        new_changes = {}
        update_previous_migration_paths = {}
        for app_label, app_migrations in changes.items():
            
            leaf_migration_nodes = loader.graph.leaf_nodes(app=app_label)
            if len(leaf_migration_nodes) == 0:
                raise CommandError(
                    f"App {app_label} has no migration, cannot update last migration."
                )
            leaf_migration_node = leaf_migration_nodes[0]
            
            leaf_migration = loader.graph.nodes[leaf_migration_node]
            
            
            if leaf_migration.replaces:
                raise CommandError(
                    f"Cannot update squash migration '{leaf_migration}'."
                )
            if leaf_migration_node in loader.applied_migrations:
                raise CommandError(
                    f"Cannot update applied migration '{leaf_migration}'."
                )
            depending_migrations = [
                migration
                for migration in loader.disk_migrations.values()
                if leaf_migration_node in migration.dependencies
            ]
            if depending_migrations:
                formatted_migrations = ", ".join(
                    [f"'{migration}'" for migration in depending_migrations]
                )
                raise CommandError(
                    f"Cannot update migration '{leaf_migration}' that migrations "
                    f"{formatted_migrations} depend on."
                )
            
            for migration in app_migrations:
                leaf_migration.operations.extend(migration.operations)

                for dependency in migration.dependencies:
                    if isinstance(dependency, SwappableTuple):
                        if settings.AUTH_USER_MODEL == dependency.setting:
                            leaf_migration.dependencies.append(
                                ("__setting__", "AUTH_USER_MODEL")
                            )
                        else:
                            leaf_migration.dependencies.append(dependency)
                    elif dependency[0] != migration.app_label:
                        leaf_migration.dependencies.append(dependency)
            
            optimizer = MigrationOptimizer()
            leaf_migration.operations = optimizer.optimize(
                leaf_migration.operations, app_label
            )
            
            previous_migration_path = MigrationWriter(leaf_migration).path
            name_fragment = self.migration_name or leaf_migration.suggest_name()
            suggested_name = leaf_migration.name[:4] + f"_{name_fragment}"
            if leaf_migration.name == suggested_name:
                new_name = leaf_migration.name + "_updated"
            else:
                new_name = suggested_name
            leaf_migration.name = new_name
            
            new_changes[app_label] = [leaf_migration]
            update_previous_migration_paths[app_label] = previous_migration_path

        self.write_migration_files(new_changes, update_previous_migration_paths)

    def write_migration_files(self, changes, update_previous_migration_paths=None):
        
        directory_created = {}
        for app_label, app_migrations in changes.items():
            if self.verbosity >= 1:
                self.log(self.style.MIGRATE_HEADING("Migrations for '%s':" % app_label))
            for migration in app_migrations:
                
                writer = MigrationWriter(migration, self.include_header)
                if self.verbosity >= 1:
                    
                    
                    migration_string = self.get_relative_path(writer.path)
                    self.log("  %s\n" % self.style.MIGRATE_LABEL(migration_string))
                    for operation in migration.operations:
                        self.log("    - %s" % operation.describe())
                    if self.scriptable:
                        self.stdout.write(migration_string)
                if not self.dry_run:
                    
                    migrations_directory = os.path.dirname(writer.path)
                    if not directory_created.get(app_label):
                        os.makedirs(migrations_directory, exist_ok=True)
                        init_path = os.path.join(migrations_directory, "__init__.py")
                        if not os.path.isfile(init_path):
                            open(init_path, "w").close()
                        
                        directory_created[app_label] = True
                    migration_string = writer.as_string()
                    with open(writer.path, "w", encoding="utf-8") as fh:
                        fh.write(migration_string)
                        self.written_files.append(writer.path)
                    if update_previous_migration_paths:
                        prev_path = update_previous_migration_paths[app_label]
                        rel_prev_path = self.get_relative_path(prev_path)
                        if writer.needs_manual_porting:
                            migration_path = self.get_relative_path(writer.path)
                            self.log(
                                self.style.WARNING(
                                    f"Updated migration {migration_path} requires "
                                    f"manual porting.\n"
                                    f"Previous migration {rel_prev_path} was kept and "
                                    f"must be deleted after porting functions manually."
                                )
                            )
                        else:
                            os.remove(prev_path)
                            self.log(f"Deleted {rel_prev_path}")
                elif self.verbosity == 3:
                    
                    
                    
                    self.log(
                        self.style.MIGRATE_HEADING(
                            "Full migrations file '%s':" % writer.filename
                        )
                    )
                    self.log(writer.as_string())
        run_formatters(self.written_files)

    @staticmethod
    def get_relative_path(path):
        try:
            migration_string = os.path.relpath(path)
        except ValueError:
            migration_string = path
        if migration_string.startswith(".."):
            migration_string = path
        return migration_string

    def handle_merge(self, loader, conflicts):
        
        if self.interactive:
            questioner = InteractiveMigrationQuestioner(prompt_output=self.log_output)
        else:
            questioner = MigrationQuestioner(defaults={"ask_merge": True})

        for app_label, migration_names in conflicts.items():
            
            
            merge_migrations = []
            for migration_name in migration_names:
                migration = loader.get_migration(app_label, migration_name)
                migration.ancestry = [
                    mig
                    for mig in loader.graph.forwards_plan((app_label, migration_name))
                    if mig[0] == migration.app_label
                ]
                merge_migrations.append(migration)

            def all_items_equal(seq):
                return all(item == seq[0] for item in seq[1:])

            merge_migrations_generations = zip(*(m.ancestry for m in merge_migrations))
            common_ancestor_count = sum(
                1
                for common_ancestor_generation in takewhile(
                    all_items_equal, merge_migrations_generations
                )
            )
            if not common_ancestor_count:
                raise ValueError(
                    "Could not find common ancestor of %s" % migration_names
                )
            
            for migration in merge_migrations:
                migration.branch = migration.ancestry[common_ancestor_count:]
                migrations_ops = (
                    loader.get_migration(node_app, node_name).operations
                    for node_app, node_name in migration.branch
                )
                migration.merged_operations = sum(migrations_ops, [])
            
            
            
            if self.verbosity > 0:
                self.log(self.style.MIGRATE_HEADING("Merging %s" % app_label))
                for migration in merge_migrations:
                    self.log(self.style.MIGRATE_LABEL("  Branch %s" % migration.name))
                    for operation in migration.merged_operations:
                        self.log("    - %s" % operation.describe())
            if questioner.ask_merge(app_label):
                
                
                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",
                    (Migration,),
                    {
                        "dependencies": [
                            (app_label, migration.name)
                            for migration in merge_migrations
                        ],
                    },
                )
                parts = ["%04i" % (biggest_number + 1)]
                if self.migration_name:
                    parts.append(self.migration_name)
                else:
                    parts.append("merge")
                    leaf_names = "_".join(
                        sorted(migration.name for migration in merge_migrations)
                    )
                    if len(leaf_names) > 47:
                        parts.append(get_migration_name_timestamp())
                    else:
                        parts.append(leaf_names)
                migration_name = "_".join(parts)
                new_migration = subclass(migration_name, app_label)
                writer = MigrationWriter(new_migration, self.include_header)

                if not self.dry_run:
                    
                    with open(writer.path, "w", encoding="utf-8") as fh:
                        fh.write(writer.as_string())
                    run_formatters([writer.path])
                    if self.verbosity > 0:
                        self.log("\nCreated new merge migration %s" % writer.path)
                        if self.scriptable:
                            self.stdout.write(writer.path)
                elif self.verbosity == 3:
                    
                    
                    
                    self.log(
                        self.style.MIGRATE_HEADING(
                            "Full merge migrations file '%s':" % writer.filename
                        )
                    )
                    self.log(writer.as_string())