load migration files from disk and their status from the database migration files are expected to live in the migrations directory of an app their names are entirely unimportant from a code perspective but will probably follow the 1234name py convention on initialization this class will scan those directories and open and read the python files looking for a class called migration which should inherit from django db migrations migration see django db migrations migration for what that looks like some migrations will be marked as replacing another set of migrations these are loaded into a separate set of migrations away from the main ones if all the migrations they replace are either unapplied or missing from disk then they are injected into the main set replacing the named migrations any dependency pointers to the replaced migrations are repointed to the new migration this does mean that this class must also talk to the database as well as to disk but this is probably fine we re already not just operating in memory return the path to the migrations module for the specified applabel and a boolean indicating if the module is specified in settings migrationmodule load the migrations from all installedapps from disk self diskmigrations self unmigratedapps set self migratedapps set for appconfig in apps getappconfigs get the migrations module directory modulename explicit self migrationsmoduleappconfig label if modulename is none self unmigratedapps addappconfig label continue wasloaded modulename in sys modules try module importmodulemodulename except modulenotfounderror as e if explicit and self ignorenomigrations or not explicit and migrationsmodulename in e name split self unmigratedapps addappconfig label continue raise else module is not a package e g migrations py if not hasattrmodule path self unmigratedapps addappconfig label continue empty directories are namespaces namespace packages have no file and don t use a list for path see https docs python org3referenceimport htmlnamespacepackages if getattrmodule file none is none and not isinstance module path list self unmigratedapps addappconfig label continue force a reload if it s already loaded tests need this if wasloaded reloadmodule self migratedapps addappconfig label migrationnames name for name ispkg in pkgutil itermodulesmodule path if not ispkg and name0 not in load migrations for migrationname in migrationnames migrationpath s s modulename migrationname try migrationmodule importmodulemigrationpath except importerror as e if bad magic number in stre raise importerror couldn t import r as it appears to be a stale pyc file migrationpath from e else raise if not hasattrmigrationmodule migration raise badmigrationerror migration s in app s has no migration class migrationname appconfig label self diskmigrations appconfig label migrationname migrationmodule migration migrationname appconfig label def getmigrationself applabel nameprefix return the migrations which match the given app label and nameprefix do the search specialcase first which means the first migration for migrated apps and is ignored for unmigrated apps it allows makemigrations to declare dependencies on apps before they even have migrations ignore first references to the same app 22325 this app isn t migrated but something depends on it the models will get autoadded into the state though so we re fine internal dependencies need to be added first to ensure first dependencies find the correct root node ignore first references to the same app skip internal dependencies build a migration dependency graph using both the disk and database you ll need to rebuild the graph if you apply migrations this isn t usually a problem as generally migration stuff runs in a oneshot process load disk data load database data to start populate the migration graph with nodes for all migrations and their dependencies also make note of replacing migrations at this step replacing migrations internal same app dependencies add external dependencies now that the internal ones have been resolved carry out replacements where possible and if enabled get applied status of each of this migration s replacement targets the replacing migration is only marked as applied if all of its replacement targets are a replacing migration can be used if either all or none of its replacement targets have been applied this replacing migration cannot be used because it is partially applied remove it from the graph and remap dependencies to it 25945 ensure the graph is consistent check if the missing node could have been replaced by any squash migration but wasn t because the squash migration was partially applied before in that case raise a more understandable exception 23556 get reverse replacements try to reraise exception with more detail raise inconsistentmigrationhistory if any applied migrations have unapplied dependencies if the migration is unknown skip it skip unapplied squashed migrations that have all of their replaces applied look through the loaded graph and detect any conflicts apps with more than one leaf migration return a dict of the app labels that conflict with the migration names that conflict return a projectstate object representing the most recent state that the loaded migrations represent see graph makestate for the meaning of nodes and atend take a migration plan and return a list of collected sql statements that represent the bestefforts version of that plan load migration files from disk and their status from the database migration files are expected to live in the migrations directory of an app their names are entirely unimportant from a code perspective but will probably follow the 1234_name py convention on initialization this class will scan those directories and open and read the python files looking for a class called migration which should inherit from django db migrations migration see django db migrations migration for what that looks like some migrations will be marked as replacing another set of migrations these are loaded into a separate set of migrations away from the main ones if all the migrations they replace are either unapplied or missing from disk then they are injected into the main set replacing the named migrations any dependency pointers to the replaced migrations are re pointed to the new migration this does mean that this class must also talk to the database as well as to disk but this is probably fine we re already not just operating in memory return the path to the migrations module for the specified app_label and a boolean indicating if the module is specified in settings migration_module load the migrations from all installed_apps from disk get the migrations module directory module is not a package e g migrations py empty directories are namespaces namespace packages have no __file__ and don t use a list for __path__ see https docs python org 3 reference import html namespace packages force a reload if it s already loaded tests need this load migrations return the named migration or raise nodenotfounderror return the migration s which match the given app label and name_prefix do the search special case __first__ which means the first migration for migrated apps and is ignored for unmigrated apps it allows makemigrations to declare dependencies on apps before they even have migrations ignore __first__ references to the same app 22325 this app isn t migrated but something depends on it the models will get auto added into the state though so we re fine __latest__ internal dependencies need to be added first to ensure __first__ dependencies find the correct root node ignore __first__ references to the same app skip internal dependencies build a migration dependency graph using both the disk and database you ll need to rebuild the graph if you apply migrations this isn t usually a problem as generally migration stuff runs in a one shot process load disk data load database data to start populate the migration graph with nodes for all migrations and their dependencies also make note of replacing migrations at this step replacing migrations internal same app dependencies add external dependencies now that the internal ones have been resolved carry out replacements where possible and if enabled get applied status of each of this migration s replacement targets the replacing migration is only marked as applied if all of its replacement targets are a replacing migration can be used if either all or none of its replacement targets have been applied this replacing migration cannot be used because it is partially applied remove it from the graph and remap dependencies to it 25945 ensure the graph is consistent check if the missing node could have been replaced by any squash migration but wasn t because the squash migration was partially applied before in that case raise a more understandable exception 23556 get reverse replacements try to reraise exception with more detail raise inconsistentmigrationhistory if any applied migrations have unapplied dependencies if the migration is unknown skip it skip unapplied squashed migrations that have all of their replaces applied look through the loaded graph and detect any conflicts apps with more than one leaf migration return a dict of the app labels that conflict with the migration names that conflict return a projectstate object representing the most recent state that the loaded migrations represent see graph make_state for the meaning of nodes and at_end take a migration plan and return a list of collected sql statements that represent the best efforts version of that plan
import pkgutil
import sys
from importlib import import_module, reload

from django.apps import apps
from django.conf import settings
from django.db.migrations.graph import MigrationGraph
from django.db.migrations.recorder import MigrationRecorder

from .exceptions import (
    AmbiguityError,
    BadMigrationError,
    InconsistentMigrationHistory,
    NodeNotFoundError,
)

MIGRATIONS_MODULE_NAME = "migrations"


class MigrationLoader:
    

    def __init__(
        self,
        connection,
        load=True,
        ignore_no_migrations=False,
        replace_migrations=True,
    ):
        self.connection = connection
        self.disk_migrations = None
        self.applied_migrations = None
        self.ignore_no_migrations = ignore_no_migrations
        self.replace_migrations = replace_migrations
        if load:
            self.build_graph()

    @classmethod
    def migrations_module(cls, app_label):
        
        if app_label in settings.MIGRATION_MODULES:
            return settings.MIGRATION_MODULES[app_label], True
        else:
            app_package_name = apps.get_app_config(app_label).name
            return "%s.%s" % (app_package_name, MIGRATIONS_MODULE_NAME), False

    def load_disk(self):
        
        self.disk_migrations = {}
        self.unmigrated_apps = set()
        self.migrated_apps = set()
        for app_config in apps.get_app_configs():
            
            module_name, explicit = self.migrations_module(app_config.label)
            if module_name is None:
                self.unmigrated_apps.add(app_config.label)
                continue
            was_loaded = module_name in sys.modules
            try:
                module = import_module(module_name)
            except ModuleNotFoundError as e:
                if (explicit and self.ignore_no_migrations) or (
                    not explicit and MIGRATIONS_MODULE_NAME in e.name.split(".")
                ):
                    self.unmigrated_apps.add(app_config.label)
                    continue
                raise
            else:
                
                if not hasattr(module, "__path__"):
                    self.unmigrated_apps.add(app_config.label)
                    continue
                
                
                
                if getattr(module, "__file__", None) is None and not isinstance(
                    module.__path__, list
                ):
                    self.unmigrated_apps.add(app_config.label)
                    continue
                
                if was_loaded:
                    reload(module)
            self.migrated_apps.add(app_config.label)
            migration_names = {
                name
                for _, name, is_pkg in pkgutil.iter_modules(module.__path__)
                if not is_pkg and name[0] not in "_~"
            }
            
            for migration_name in migration_names:
                migration_path = "%s.%s" % (module_name, migration_name)
                try:
                    migration_module = import_module(migration_path)
                except ImportError as e:
                    if "bad magic number" in str(e):
                        raise ImportError(
                            "Couldn't import %r as it appears to be a stale "
                            ".pyc file." % migration_path
                        ) from e
                    else:
                        raise
                if not hasattr(migration_module, "Migration"):
                    raise BadMigrationError(
                        "Migration %s in app %s has no Migration class"
                        % (migration_name, app_config.label)
                    )
                self.disk_migrations[
                    app_config.label, migration_name
                ] = migration_module.Migration(
                    migration_name,
                    app_config.label,
                )

    def get_migration(self, app_label, name_prefix):
        
        return self.graph.nodes[app_label, name_prefix]

    def get_migration_by_prefix(self, app_label, name_prefix):
        
        
        results = []
        for migration_app_label, migration_name in self.disk_migrations:
            if migration_app_label == app_label and migration_name.startswith(
                name_prefix
            ):
                results.append((migration_app_label, migration_name))
        if len(results) > 1:
            raise AmbiguityError(
                "There is more than one migration for '%s' with the prefix '%s'"
                % (app_label, name_prefix)
            )
        elif not results:
            raise KeyError(
                f"There is no migration for '{app_label}' with the prefix "
                f"'{name_prefix}'"
            )
        else:
            return self.disk_migrations[results[0]]

    def check_key(self, key, current_app):
        if (key[1] != "__first__" and key[1] != "__latest__") or key in self.graph:
            return key
        
        
        
        
        if key[0] == current_app:
            
            return
        if key[0] in self.unmigrated_apps:
            
            
            
            return
        if key[0] in self.migrated_apps:
            try:
                if key[1] == "__first__":
                    return self.graph.root_nodes(key[0])[0]
                else:  
                    return self.graph.leaf_nodes(key[0])[0]
            except IndexError:
                if self.ignore_no_migrations:
                    return None
                else:
                    raise ValueError(
                        "Dependency on app with no migrations: %s" % key[0]
                    )
        raise ValueError("Dependency on unknown app: %s" % key[0])

    def add_internal_dependencies(self, key, migration):
        
        for parent in migration.dependencies:
            
            if parent[0] == key[0] and parent[1] != "__first__":
                self.graph.add_dependency(migration, key, parent, skip_validation=True)

    def add_external_dependencies(self, key, migration):
        for parent in migration.dependencies:
            
            if key[0] == parent[0]:
                continue
            parent = self.check_key(parent, key[0])
            if parent is not None:
                self.graph.add_dependency(migration, key, parent, skip_validation=True)
        for child in migration.run_before:
            child = self.check_key(child, key[0])
            if child is not None:
                self.graph.add_dependency(migration, child, key, skip_validation=True)

    def build_graph(self):
        
        
        self.load_disk()
        
        if self.connection is None:
            self.applied_migrations = {}
        else:
            recorder = MigrationRecorder(self.connection)
            self.applied_migrations = recorder.applied_migrations()
        
        
        self.graph = MigrationGraph()
        self.replacements = {}
        for key, migration in self.disk_migrations.items():
            self.graph.add_node(key, migration)
            
            if migration.replaces:
                self.replacements[key] = migration
        for key, migration in self.disk_migrations.items():
            
            self.add_internal_dependencies(key, migration)
        
        for key, migration in self.disk_migrations.items():
            self.add_external_dependencies(key, migration)
        
        if self.replace_migrations:
            for key, migration in self.replacements.items():
                
                
                applied_statuses = [
                    (target in self.applied_migrations) for target in migration.replaces
                ]
                
                
                if all(applied_statuses):
                    self.applied_migrations[key] = migration
                else:
                    self.applied_migrations.pop(key, None)
                
                
                if all(applied_statuses) or (not any(applied_statuses)):
                    self.graph.remove_replaced_nodes(key, migration.replaces)
                else:
                    
                    
                    
                    self.graph.remove_replacement_node(key, migration.replaces)
        
        try:
            self.graph.validate_consistency()
        except NodeNotFoundError as exc:
            
            
            
            
            
            reverse_replacements = {}
            for key, migration in self.replacements.items():
                for replaced in migration.replaces:
                    reverse_replacements.setdefault(replaced, set()).add(key)
            
            if exc.node in reverse_replacements:
                candidates = reverse_replacements.get(exc.node, set())
                is_replaced = any(
                    candidate in self.graph.nodes for candidate in candidates
                )
                if not is_replaced:
                    tries = ", ".join("%s.%s" % c for c in candidates)
                    raise NodeNotFoundError(
                        "Migration {0} depends on nonexistent node ('{1}', '{2}'). "
                        "Django tried to replace migration {1}.{2} with any of [{3}] "
                        "but wasn't able to because some of the replaced migrations "
                        "are already applied.".format(
                            exc.origin, exc.node[0], exc.node[1], tries
                        ),
                        exc.node,
                    ) from exc
            raise
        self.graph.ensure_not_cyclic()

    def check_consistent_history(self, connection):
        
        recorder = MigrationRecorder(connection)
        applied = recorder.applied_migrations()
        for migration in applied:
            
            if migration not in self.graph.nodes:
                continue
            for parent in self.graph.node_map[migration].parents:
                if parent not in applied:
                    
                    
                    if parent in self.replacements:
                        if all(
                            m in applied for m in self.replacements[parent].replaces
                        ):
                            continue
                    raise InconsistentMigrationHistory(
                        "Migration {}.{} is applied before its dependency "
                        "{}.{} on database '{}'.".format(
                            migration[0],
                            migration[1],
                            parent[0],
                            parent[1],
                            connection.alias,
                        )
                    )

    def detect_conflicts(self):
        
        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 {
            app_label: sorted(seen_apps[app_label]) for app_label in conflicting_apps
        }

    def project_state(self, nodes=None, at_end=True):
        
        return self.graph.make_state(
            nodes=nodes, at_end=at_end, real_apps=self.unmigrated_apps
        )

    def collect_sql(self, plan):
        
        statements = []
        state = None
        for migration, backwards in plan:
            with self.connection.schema_editor(
                collect_sql=True, atomic=migration.atomic
            ) as schema_editor:
                if state is None:
                    state = self.project_state(
                        (migration.app_label, migration.name), at_end=False
                    )
                if not backwards:
                    state = migration.apply(state, schema_editor, collect_sql=True)
                else:
                    state = migration.unapply(state, schema_editor, collect_sql=True)
            statements.extend(schema_editor.collected_sql)
        return statements