Adds command to check for circular migration dependencies

django_squash/db/migrations/autodetector.py

@@ -10,7 +10,8 @@
 from django.db.migrations.autodetector import MigrationAutodetector as MigrationAutodetectorBase
 
 from django_squash.contrib import postgres
-from django_squash.db.migrations import utils
+
+from . import utils
 
 RESERVED_MIGRATION_KEYWORDS = ("_deleted", "_dependencies_change", "_replaces_change", "_original_migration")
 
@@ -83,13 +84,10 @@
                         continue
 
                     if isinstance(operation, dj_migrations.RunSQL):
-                        operation._original_migration = migration
                         new_operations.append(operation)
                     elif isinstance(operation, dj_migrations.RunPython):
-                        operation._original_migration = migration
                         new_operations.append(operation)
                     elif isinstance(operation, postgres.PGCreateExtension):
-                        operation._original_migration = migration
                         new_operations_bubble_top.append(operation)
                     elif isinstance(operation, dj_migrations.SeparateDatabaseAndState):
                         # A valid use case for this should be given before any work is done.
@@ -160,7 +158,7 @@
                 for dependency in migration.dependencies:
                     dep_app_label, dep_migration = dependency
                     if dep_app_label in ignore_apps:
-                        new_dependencies.append(original.graph.leaf_nodes(dep_app_label)[0])
+                        new_dependencies.append(original.graph.leaf_nodes(dependency[0])[0])
                         continue
 
                     if dep_app_label == "__setting__":
@@ -174,10 +172,10 @@
                             # Leave as is, the django's migration writer will handle this by default
                             new_dependencies.append(dependency)
                             continue
-                    # Technically, the terms '__first__' and '__latest__' could apply to dependencies. However, these
-                    # are not labels that Django assigns automatically. Instead, they would be manually specified by
-                    # the developer after Django has generated the necessary files. Given that our focus is solely
-                    # on handling migrations created by Django, there is no practical need to account for these.
+                    elif dep_migration == "__first__":
+                        dependency = original.graph.root_nodes(dep_app_label)[0]
+                    elif dep_migration == "__latest__":
+                        dependency = original.graph.leaf_nodes(dep_app_label)[0]
 
                     migration_id = dependency
                     if migration_id not in migrations_by_name:

django_squash/db/migrations/loader.py

@@ -1,4 +1,4 @@
-import logging
+import inspect
 import os
 import tempfile
 from contextlib import ExitStack
@@ -7,38 +7,30 @@
 from django.conf import settings
 from django.db.migrations.loader import MigrationLoader
 
-from django_squash.db.migrations import utils
-
-logger = logging.getLogger(__name__)
-
 
 class SquashMigrationLoader(MigrationLoader):
     def __init__(self, *args, **kwargs):
         # keep a copy of the original migration modules to restore it later
         original_migration_modules = settings.MIGRATION_MODULES
         # make a copy of the migration modules so we can modify it
         settings.MIGRATION_MODULES = settings.MIGRATION_MODULES.copy()
-        site_packages_path = utils.site_packages_path()
+        project_path = os.path.abspath(os.curdir)
 
         with ExitStack() as stack:
             # Find each app that belongs to the user and are not in the site-packages. Create a fake temporary
             # directory inside each app that will tell django we don't have any migrations at all.
             for app_config in apps.get_app_configs():
-                # absolute path to the app
-                app_path = utils.source_directory(app_config.module)
-
-                if app_path.startswith(site_packages_path):
-                    # ignore any apps in inside site-packages
-                    logger.debug("Ignoring app %s inside site-packages: %s", app_config.label, app_path)
-                    continue
-
-                temp_dir = stack.enter_context(tempfile.TemporaryDirectory(prefix="migrations_", dir=app_path))
-                # Need to make this directory a proper python module otherwise django will refuse to recognize it
-                open(os.path.join(temp_dir, "__init__.py"), "a").close()
-                settings.MIGRATION_MODULES[app_config.label] = "%s.%s" % (
-                    app_config.module.__name__,
-                    os.path.basename(temp_dir),
-                )
+                module = app_config.module
+                app_path = os.path.dirname(os.path.abspath(inspect.getsourcefile(module)))
+
+                if app_path.startswith(project_path):
+                    temp_dir = stack.enter_context(tempfile.TemporaryDirectory(prefix="migrations_", dir=app_path))
+                    # Need to make this directory a proper python module otherwise django will refuse to recognize it
+                    open(os.path.join(temp_dir, "__init__.py"), "a").close()
+                    settings.MIGRATION_MODULES[app_config.label] = "%s.%s" % (
+                        app_config.module.__name__,
+                        os.path.basename(temp_dir),
+                    )
 
             super().__init__(*args, **kwargs)
 

django_squash/db/migrations/utils.py

@@ -1,5 +1,4 @@
 import ast
-import functools
 import hashlib
 import importlib
 import inspect
@@ -9,6 +8,7 @@
 import sysconfig
 import types
 from collections import defaultdict
+import functools
 
 from django.db import migrations
 from django.utils.module_loading import import_string
@@ -45,9 +45,6 @@
 
 
 def source_directory(module):
-    """
-    Return the absolute path of a module
-    """
     return os.path.dirname(os.path.abspath(inspect.getsourcefile(module)))
 
 
@@ -75,37 +72,39 @@
         if inspect.ismethod(func) or inspect.signature(func).parameters.get("self") is not None:
             raise ValueError("func cannot be part of an instance")
 
-        name = func.__qualname__
+        name = original_name = func.__qualname__
         if "." in name:
             parent_name, actual_name = name.rsplit(".", 1)
             parent = getattr(import_string(func.__module__), parent_name)
             if issubclass(parent, migrations.Migration):
-                name = actual_name
-
-        if func in self.functions:
+                name = name = original_name = actual_name
+        already_accounted = func in self.functions
+        if already_accounted:
             return self.functions[func]
 
-        name = self.naming_function(name, {**self.context, "type_": "function", "func": func})
-        new_name = self.functions[func] = self.uniq(name)
-
-        return new_name
-
-    def uniq(self, name, original_name=None):
-        original_name = original_name or name
         # Endless loop that will try different combinations until it finds a unique name
         for i, _ in enumerate(itertools.count(), 2):
             if self.names[name] == 0:
+                self.functions[func] = name
                 self.names[name] += 1
                 break
 
             name = "%s_%s" % (original_name, i)
+
+        self.functions[func] = name
+
         return name
 
     def __call__(self, name, force_number=False):
-        original_name = name
-        if force_number:
-            name = f"{name}_1"
-        return self.uniq(name, original_name)
+        self.names[name] += 1
+        count = self.names[name]
+        if not force_number and count == 1:
+            return name
+        else:
+            new_name = "%s_%s" % (name, count)
+            # Make sure that the function name is fully unique
+            # You can potentially have the same name already defined.
+            return self(new_name)
 
 
 def get_imports(module):
@@ -130,10 +129,9 @@
 
 
 def normalize_function_name(name):
-    _, _, function_name = name.rpartition(".")
-    if function_name[0].isdigit():
-        # Functions CANNOT start with a number
-        function_name = "f_" + function_name
+    class_name, _, function_name = name.rpartition(".")
+    if class_name and not function_name:
+        function_name = class_name
     return function_name
 
 
@@ -145,10 +143,10 @@
     func.__qualname__ = name
     func.__original__ = f
     func.__source__ = re.sub(
-        pattern=rf"(def\s+){normalize_function_name(f.__qualname__)}",
-        repl=rf"\1{name}",
-        string=inspect.getsource(f),
-        count=1,
+        rf"(def\s+){normalize_function_name(f.__qualname__)}",
+        rf"\1{normalize_function_name(name)}",
+        inspect.getsource(f),
+        1,
     )
     return func
 
@@ -168,18 +166,12 @@
 
 def is_code_in_site_packages(module_name):
     # Find the module in the site-packages directory
-    site_packages_path_ = site_packages_path()
+    site_packages_path = sysconfig.get_path("purelib")  # returns the "../site-packages" directory
     try:
         loader = importlib.util.find_spec(module_name)
+        return site_packages_path in loader.origin
     except ImportError:
         return False
-    return loader.origin.startswith(site_packages_path_)
-
-
-@functools.lru_cache(maxsize=1)
-def site_packages_path():
-    # returns the "../site-packages" directory
-    return sysconfig.get_path("purelib")
 
 
 def replace_migration_attribute(source, attr, value):

django_squash/db/migrations/writer.py

@@ -166,20 +166,8 @@
             return self.replace_in_migration()
 
         variables = []
-        custom_naming_function = utils.get_custom_rename_function()
-        unique_names = utils.UniqueVariableName(
-            {"app": self.migration.app_label}, naming_function=custom_naming_function
-        )
+        unique_names = utils.UniqueVariableName()
         for operation in self.migration.operations:
-            unique_names.update_context(
-                {
-                    "new_migration": self.migration,
-                    "operation": operation,
-                    "migration": (
-                        operation._original_migration if hasattr(operation, "_original_migration") else self.migration
-                    ),
-                }
-            )
             operation._deconstruct = operation.__class__.deconstruct
 
             def deconstruct(self):
@@ -191,14 +179,12 @@
                 # Bind the deconstruct() to the instance to get the elidable
                 operation.deconstruct = deconstruct.__get__(operation, operation.__class__)
                 if not utils.is_code_in_site_packages(operation.code.__module__):
-                    code_name = utils.normalize_function_name(unique_names.function(operation.code))
+                    code_name = unique_names.function(operation.code)
                     operation.code = utils.copy_func(operation.code, code_name)
                     operation.code.__in_migration_file__ = True
                 if operation.reverse_code:
                     if not utils.is_code_in_site_packages(operation.reverse_code.__module__):
-                        reversed_code_name = utils.normalize_function_name(
-                            unique_names.function(operation.reverse_code)
-                        )
+                        reversed_code_name = unique_names.function(operation.reverse_code)
                         operation.reverse_code = utils.copy_func(operation.reverse_code, reversed_code_name)
                         operation.reverse_code.__in_migration_file__ = True
             elif isinstance(operation, dj_migrations.RunSQL):

django_squash/management/commands/circularmigrations.py

@@ -0,0 +1,86 @@
+from django.apps import apps
+from django.conf import settings
+from django.core.management.base import BaseCommand, no_translations, CommandError
+from django.db.migrations.state import ProjectState
+
+from django_squash.db.migrations.autodetector import SquashMigrationAutodetector
+from django_squash.db.migrations.loader import SquashMigrationLoader
+from django_squash.db.migrations.questioner import NonInteractiveMigrationQuestioner
+
+
+class Command(BaseCommand):
+    def add_arguments(self, parser):
+        pass
+
+    @no_translations
+    def handle(self, **kwargs):
+        self.verbosity = 1
+        self.include_header = False
+
+        dependency_list = settings.INSTALLED_APPS
+        apps.all_models
+
+        questioner = NonInteractiveMigrationQuestioner(specified_apps=None, dry_run=False)
+        squash_loader = SquashMigrationLoader(None, ignore_no_migrations=True)
+        autodetector = SquashMigrationAutodetector(
+            squash_loader.project_state(),
+            ProjectState.from_apps(apps),
+            questioner,
+        )
+
+        changes = autodetector.changes(
+            squash_loader.graph,
+            trim_to_apps=None,
+            convert_apps=None,
+            migration_name=None,
+        )
+        for app_label, migrations in changes.items():
+            app_label = apps.get_app_config(app_label).name
+            try:
+                app_ranking = dependency_list.index(app_label)
+            except ValueError:
+                print(f"{app_label} not found")
+                continue
+            for migration in migrations:
+                bad_dependencies = []
+                for depends_on_app_label, _ in migration.dependencies:
+                    if depends_on_app_label == "__setting__":
+                        continue
+                    depends_on_app_label = apps.get_app_config(depends_on_app_label).name
+                    depends_on_app_ranking = dependency_list.index(depends_on_app_label)
+                    if depends_on_app_ranking > app_ranking:
+                        print(f"* {app_label} ({app_ranking}) > {depends_on_app_label} ({depends_on_app_ranking})")
+                        bad_dependencies.append(depends_on_app_label.split(".")[-1])
+
+                if not bad_dependencies:
+                    continue
+
+                references_found = []
+                for operation in migration.operations:
+                    if hasattr(operation, "field") and hasattr(operation.field, "related_model"):
+                        if get_related_model(operation.field.related_model)._meta.app_label in bad_dependencies:
+                            references_found.append(
+                                (
+                                    operation.model_name,
+                                    operation.name,
+                                    get_related_model(operation.field.related_model),
+                                )
+                            )
+
+                    for name, field in getattr(operation, "fields", []):
+                        if (hasattr(field, "related_model")):
+                            if get_related_model(field.related_model)._meta.app_label in bad_dependencies:
+                                references_found.append((operation.name, name, get_related_model(field.related_model)))
+
+                for model_name, field_name, model in references_found:
+                    print(
+                        f"  references {model._meta.app_label}.{model._meta.object_name} via {model_name}.{field_name}"
+                    )
+                if references_found:
+                    raise CommandError("Bad dependencies found")
+
+
+def get_related_model(model):
+    if isinstance(model, str):
+        return apps.get_model(model)
+    return model