endtoend migration execution load migrations and run them up or down to a specified set of targets given a set of targets return a list of migration instance backwards if the target is applabel none that means unmigrate everything if the migration is already applied do backwards mode otherwise do forwards mode if the target is missing it s likely a replaced migration reload the graph without replacements don t migrate backwards all the way to the target node that may roll back dependencies in other apps that don t need to be rolled back instead roll back through target s immediate children in the same app and no further create a project state including all the applications without migrations and applied migrations if withappliedmigrationstrue create the forwards plan django would follow on an empty database migrate the database up to the given targets django first needs to create all project states before a migration is unapplied and in a second step run all the database operations the djangomigrations table must be present to record applied migrations but don t create it if there are no migrations to apply create the forwards plan django would follow on an empty database the resulting state should include applied migrations this should only happen if there s a mixed plan the resulting state should still include applied migrations no need to check for elif allbackwards here as that condition would always evaluate to true take a list of 2tuples of the form migration instance false and apply them in the order they occur in the fullplan we remove every migration that we applied from these sets so that we can bail out once the last migration has been applied and don t always run until the very end of the migration process take a list of 2tuples of the form migration instance true and unapply them in reverse order they occur in the fullplan since unapplying a migration requires the project state prior to that migration django will compute the migration states before each of them in a first run over the plan and then unapply them in a second run over the plan holds all migration states prior to the migrations being unapplied we remove every migration that we applied from this set so that we can bail out once the last migration has been applied and don t always run until the very end of the migration process the state before this migration the old state keeps asis we continue with the new state only mutate the state if the migration is actually applied to make sure the resulting state doesn t include changes from unrelated migrations generate the post migration state by starting from the state before the last migration is unapplied and mutating it to include all the remaining applied migrations run a migration forwards migrationrecorded false if self progresscallback self progresscallbackapplystart migration fake if not fake if fakeinitial test to see if this is an alreadyapplied initial migration applied state self detectsoftappliedstate migration if applied fake true if not fake alright do it normally with self connection schemaeditor atomicmigration atomic as schemaeditor state migration applystate schemaeditor if not schemaeditor deferredsql self recordmigrationmigration migrationrecorded true if not migrationrecorded self recordmigrationmigration report progress if self progresscallback self progresscallbackapplysuccess migration fake return state def recordmigrationself migration for replacement migrations record individual statuses if migration replaces for applabel name in migration replaces self recorder recordappliedapplabel name else self recorder recordappliedmigration applabel migration name def unapplymigrationself state migration fakefalse for replacement migrations also record individual statuses report progress mark replacement migrations applied if their replaced set all are do this unconditionally on every migrate rather than just when migrations are applied or unapplied to correctly handle the case when a new squash migration is pushed to a deployment that already had all its replaced migrations applied in this case no new migration will be applied but the applied state of the squashed migration must be maintained test whether a migration has been implicitly applied that the tables or columns it would create exist this is intended only for use on initial migrations as it only looks for createmodel and addfield no need to detect tables for proxy models unmanaged models or models that can t be migrated on the current database bail if the migration isn t the first one in its app bail if it s not an initial migration make sure all create model and add field operations are done we have to fetch the model to test with from the main app cache as it s not a direct dependency we have to fetch the model to test with from the main app cache as it s not a direct dependency handle implicit manytomany tables addfield if we get this far and we found at least one createmodel or addfield migration the migration is considered implicitly applied end to end migration execution load migrations and run them up or down to a specified set of targets given a set of targets return a list of migration instance backwards if the target is app_label none that means unmigrate everything if the migration is already applied do backwards mode otherwise do forwards mode if the target is missing it s likely a replaced migration reload the graph without replacements don t migrate backwards all the way to the target node that may roll back dependencies in other apps that don t need to be rolled back instead roll back through target s immediate child ren in the same app and no further create a project state including all the applications without migrations and applied migrations if with_applied_migrations true create the forwards plan django would follow on an empty database migrate the database up to the given targets django first needs to create all project states before a migration is un applied and in a second step run all the database operations the django_migrations table must be present to record applied migrations but don t create it if there are no migrations to apply create the forwards plan django would follow on an empty database the resulting state should include applied migrations this should only happen if there s a mixed plan the resulting state should still include applied migrations no need to check for elif all_backwards here as that condition would always evaluate to true take a list of 2 tuples of the form migration instance false and apply them in the order they occur in the full_plan we remove every migration that we applied from these sets so that we can bail out once the last migration has been applied and don t always run until the very end of the migration process render all performance critical take a list of 2 tuples of the form migration instance true and unapply them in reverse order they occur in the full_plan since unapplying a migration requires the project state prior to that migration django will compute the migration states before each of them in a first run over the plan and then unapply them in a second run over the plan holds all migration states prior to the migrations being unapplied we remove every migration that we applied from this set so that we can bail out once the last migration has been applied and don t always run until the very end of the migration process render all performance critical the state before this migration the old state keeps as is we continue with the new state only mutate the state if the migration is actually applied to make sure the resulting state doesn t include changes from unrelated migrations generate the post migration state by starting from the state before the last migration is unapplied and mutating it to include all the remaining applied migrations run a migration forwards test to see if this is an already applied initial migration alright do it normally report progress for replacement migrations record individual statuses run a migration backwards for replacement migrations also record individual statuses report progress mark replacement migrations applied if their replaced set all are do this unconditionally on every migrate rather than just when migrations are applied or unapplied to correctly handle the case when a new squash migration is pushed to a deployment that already had all its replaced migrations applied in this case no new migration will be applied but the applied state of the squashed migration must be maintained test whether a migration has been implicitly applied that the tables or columns it would create exist this is intended only for use on initial migrations as it only looks for createmodel and addfield no need to detect tables for proxy models unmanaged models or models that can t be migrated on the current database bail if the migration isn t the first one in its app bail if it s not an initial migration make sure all create model and add field operations are done we have to fetch the model to test with from the main app cache as it s not a direct dependency we have to fetch the model to test with from the main app cache as it s not a direct dependency handle implicit many to many tables addfield if we get this far and we found at least one createmodel or addfield migration the migration is considered implicitly applied
from django.apps.registry import apps as global_apps
from django.db import migrations, router

from .exceptions import InvalidMigrationPlan
from .loader import MigrationLoader
from .recorder import MigrationRecorder
from .state import ProjectState


class MigrationExecutor:
    

    def __init__(self, connection, progress_callback=None):
        self.connection = connection
        self.loader = MigrationLoader(self.connection)
        self.recorder = MigrationRecorder(self.connection)
        self.progress_callback = progress_callback

    def migration_plan(self, targets, clean_start=False):
        
        plan = []
        if clean_start:
            applied = {}
        else:
            applied = dict(self.loader.applied_migrations)
        for target in targets:
            
            if target[1] is None:
                for root in self.loader.graph.root_nodes():
                    if root[0] == target[0]:
                        for migration in self.loader.graph.backwards_plan(root):
                            if migration in applied:
                                plan.append((self.loader.graph.nodes[migration], True))
                                applied.pop(migration)
            
            
            elif target in applied:
                
                
                if (
                    self.loader.replace_migrations
                    and target not in self.loader.graph.node_map
                ):
                    self.loader.replace_migrations = False
                    self.loader.build_graph()
                    return self.migration_plan(targets, clean_start=clean_start)
                
                
                
                
                next_in_app = sorted(
                    n
                    for n in self.loader.graph.node_map[target].children
                    if n[0] == target[0]
                )
                for node in next_in_app:
                    for migration in self.loader.graph.backwards_plan(node):
                        if migration in applied:
                            plan.append((self.loader.graph.nodes[migration], True))
                            applied.pop(migration)
            else:
                for migration in self.loader.graph.forwards_plan(target):
                    if migration not in applied:
                        plan.append((self.loader.graph.nodes[migration], False))
                        applied[migration] = self.loader.graph.nodes[migration]
        return plan

    def _create_project_state(self, with_applied_migrations=False):
        
        state = ProjectState(real_apps=self.loader.unmigrated_apps)
        if with_applied_migrations:
            
            full_plan = self.migration_plan(
                self.loader.graph.leaf_nodes(), clean_start=True
            )
            applied_migrations = {
                self.loader.graph.nodes[key]
                for key in self.loader.applied_migrations
                if key in self.loader.graph.nodes
            }
            for migration, _ in full_plan:
                if migration in applied_migrations:
                    migration.mutate_state(state, preserve=False)
        return state

    def migrate(self, targets, plan=None, state=None, fake=False, fake_initial=False):
        
        
        
        if plan == []:
            if not self.recorder.has_table():
                return self._create_project_state(with_applied_migrations=False)
        else:
            self.recorder.ensure_schema()

        if plan is None:
            plan = self.migration_plan(targets)
        
        full_plan = self.migration_plan(
            self.loader.graph.leaf_nodes(), clean_start=True
        )

        all_forwards = all(not backwards for mig, backwards in plan)
        all_backwards = all(backwards for mig, backwards in plan)

        if not plan:
            if state is None:
                
                state = self._create_project_state(with_applied_migrations=True)
        elif all_forwards == all_backwards:
            
            raise InvalidMigrationPlan(
                "Migration plans with both forwards and backwards migrations "
                "are not supported. Please split your migration process into "
                "separate plans of only forwards OR backwards migrations.",
                plan,
            )
        elif all_forwards:
            if state is None:
                
                state = self._create_project_state(with_applied_migrations=True)
            state = self._migrate_all_forwards(
                state, plan, full_plan, fake=fake, fake_initial=fake_initial
            )
        else:
            
            
            state = self._migrate_all_backwards(plan, full_plan, fake=fake)

        self.check_replacements()

        return state

    def _migrate_all_forwards(self, state, plan, full_plan, fake, fake_initial):
        
        migrations_to_run = {m[0] for m in plan}
        for migration, _ in full_plan:
            if not migrations_to_run:
                
                
                
                
                break
            if migration in migrations_to_run:
                if "apps" not in state.__dict__:
                    if self.progress_callback:
                        self.progress_callback("render_start")
                    state.apps  
                    if self.progress_callback:
                        self.progress_callback("render_success")
                state = self.apply_migration(
                    state, migration, fake=fake, fake_initial=fake_initial
                )
                migrations_to_run.remove(migration)

        return state

    def _migrate_all_backwards(self, plan, full_plan, fake):
        
        migrations_to_run = {m[0] for m in plan}
        
        states = {}
        state = self._create_project_state()
        applied_migrations = {
            self.loader.graph.nodes[key]
            for key in self.loader.applied_migrations
            if key in self.loader.graph.nodes
        }
        if self.progress_callback:
            self.progress_callback("render_start")
        for migration, _ in full_plan:
            if not migrations_to_run:
                
                
                
                
                break
            if migration in migrations_to_run:
                if "apps" not in state.__dict__:
                    state.apps  
                
                states[migration] = state
                
                state = migration.mutate_state(state, preserve=True)
                migrations_to_run.remove(migration)
            elif migration in applied_migrations:
                
                
                
                migration.mutate_state(state, preserve=False)
        if self.progress_callback:
            self.progress_callback("render_success")

        for migration, _ in plan:
            self.unapply_migration(states[migration], migration, fake=fake)
            applied_migrations.remove(migration)

        
        
        
        last_unapplied_migration = plan[-1][0]
        state = states[last_unapplied_migration]
        for index, (migration, _) in enumerate(full_plan):
            if migration == last_unapplied_migration:
                for migration, _ in full_plan[index:]:
                    if migration in applied_migrations:
                        migration.mutate_state(state, preserve=False)
                break

        return state

    def apply_migration(self, state, migration, fake=False, fake_initial=False):
        
        migration_recorded = False
        if self.progress_callback:
            self.progress_callback("apply_start", migration, fake)
        if not fake:
            if fake_initial:
                
                applied, state = self.detect_soft_applied(state, migration)
                if applied:
                    fake = True
            if not fake:
                
                with self.connection.schema_editor(
                    atomic=migration.atomic
                ) as schema_editor:
                    state = migration.apply(state, schema_editor)
                    if not schema_editor.deferred_sql:
                        self.record_migration(migration)
                        migration_recorded = True
        if not migration_recorded:
            self.record_migration(migration)
        
        if self.progress_callback:
            self.progress_callback("apply_success", migration, fake)
        return state

    def record_migration(self, migration):
        
        if migration.replaces:
            for app_label, name in migration.replaces:
                self.recorder.record_applied(app_label, name)
        else:
            self.recorder.record_applied(migration.app_label, migration.name)

    def unapply_migration(self, state, migration, fake=False):
        
        if self.progress_callback:
            self.progress_callback("unapply_start", migration, fake)
        if not fake:
            with self.connection.schema_editor(
                atomic=migration.atomic
            ) as schema_editor:
                state = migration.unapply(state, schema_editor)
        
        if migration.replaces:
            for app_label, name in migration.replaces:
                self.recorder.record_unapplied(app_label, name)
        self.recorder.record_unapplied(migration.app_label, migration.name)
        
        if self.progress_callback:
            self.progress_callback("unapply_success", migration, fake)
        return state

    def check_replacements(self):
        
        applied = self.recorder.applied_migrations()
        for key, migration in self.loader.replacements.items():
            all_applied = all(m in applied for m in migration.replaces)
            if all_applied and key not in applied:
                self.recorder.record_applied(*key)

    def detect_soft_applied(self, project_state, migration):
        

        def should_skip_detecting_model(migration, model):
            
            return (
                model._meta.proxy
                or not model._meta.managed
                or not router.allow_migrate(
                    self.connection.alias,
                    migration.app_label,
                    model_name=model._meta.model_name,
                )
            )

        if migration.initial is None:
            
            if any(app == migration.app_label for app, name in migration.dependencies):
                return False, project_state
        elif migration.initial is False:
            
            return False, project_state

        if project_state is None:
            after_state = self.loader.project_state(
                (migration.app_label, migration.name), at_end=True
            )
        else:
            after_state = migration.mutate_state(project_state)
        apps = after_state.apps
        found_create_model_migration = False
        found_add_field_migration = False
        fold_identifier_case = self.connection.features.ignores_table_name_case
        with self.connection.cursor() as cursor:
            existing_table_names = set(
                self.connection.introspection.table_names(cursor)
            )
            if fold_identifier_case:
                existing_table_names = {
                    name.casefold() for name in existing_table_names
                }
        
        for operation in migration.operations:
            if isinstance(operation, migrations.CreateModel):
                model = apps.get_model(migration.app_label, operation.name)
                if model._meta.swapped:
                    
                    
                    model = global_apps.get_model(model._meta.swapped)
                if should_skip_detecting_model(migration, model):
                    continue
                db_table = model._meta.db_table
                if fold_identifier_case:
                    db_table = db_table.casefold()
                if db_table not in existing_table_names:
                    return False, project_state
                found_create_model_migration = True
            elif isinstance(operation, migrations.AddField):
                model = apps.get_model(migration.app_label, operation.model_name)
                if model._meta.swapped:
                    
                    
                    model = global_apps.get_model(model._meta.swapped)
                if should_skip_detecting_model(migration, model):
                    continue

                table = model._meta.db_table
                field = model._meta.get_field(operation.name)

                
                if field.many_to_many:
                    through_db_table = field.remote_field.through._meta.db_table
                    if fold_identifier_case:
                        through_db_table = through_db_table.casefold()
                    if through_db_table not in existing_table_names:
                        return False, project_state
                    else:
                        found_add_field_migration = True
                        continue
                with self.connection.cursor() as cursor:
                    columns = self.connection.introspection.get_table_description(
                        cursor, table
                    )
                for column in columns:
                    field_column = field.column
                    column_name = column.name
                    if fold_identifier_case:
                        column_name = column_name.casefold()
                        field_column = field_column.casefold()
                    if column_name == field_column:
                        found_add_field_migration = True
                        break
                else:
                    return False, project_state
        
        
        return (found_create_model_migration or found_add_field_migration), after_state