Introduce automated makemigrations stage splitting.

syzygy/autodetector.py

@@ -9,6 +9,7 @@
 from django.db.models.fields import NOT_PROVIDED
 
 from .operations import AddField, PostAddField, PreRemoveField
+from .plan import partition_operations
 
 
 class Stage(Operation):
@@ -106,6 +107,20 @@ def check_dependency(self, operation, dependency):
         return super().check_dependency(operation, dependency)
 
     def _build_migration_list(self, *args, **kwargs):
+        # Ensure generated operations sequence for each apps are partitioned
+        # by stage.
+        for app_label, app_operations in list(self.generated_operations.items()):
+            if app_label == self.STAGE_SPLIT:
+                continue
+            pre_operations, post_operations = partition_operations(app_operations)
+            if pre_operations and post_operations:
+                stage = Stage()
+                self.add_operation(
+                    self.STAGE_SPLIT,
+                    stage,
+                    dependencies=[(app_label, self.STAGE_SPLIT, pre_operations[-1])],
+                )
+                post_operations[0]._auto_deps.append((self.STAGE_SPLIT, stage))
         super()._build_migration_list(*args, **kwargs)
         # Remove all dangling references to stage migrations.
         if self.migrations.pop(self.STAGE_SPLIT, None):

syzygy/plan.py

@@ -24,6 +24,32 @@ def get_operation_stage(operation: Operation) -> Stage:
     return Stage.PRE_DEPLOY
 
 
+def partition_operations(
+    operations: List[Operation],
+) -> Tuple[List[Operation], List[Operation]]:
+    """
+    Partition an ordered list of operations by :class:`syzygy.constants.Stage`.
+
+    If `operations` is composed of members with a :attr:`syzygy.constants.Stage.PRE_DEPLOY`
+    stage after members with a :attr:`syzygy.constants.Stage.PRE_DEPLOY` stage
+    a :class:`syzygy.exceptions.AmbiguousStage` exception will be raised.
+    """
+    stage_operations: Dict[Stage, List[Operation]] = {
+        Stage.PRE_DEPLOY: [],
+        Stage.POST_DEPLOY: [],
+    }
+    current_stage = Stage.PRE_DEPLOY
+    for operation in operations:
+        operation_stage = get_operation_stage(operation)
+        if operation_stage is Stage.PRE_DEPLOY and current_stage is Stage.POST_DEPLOY:
+            raise AmbiguousStage(
+                "Post-deployment operations cannot be followed by pre-deployments operations"
+            )
+        stage_operations[operation_stage].append(operation)
+        current_stage = operation_stage
+    return stage_operations[Stage.PRE_DEPLOY], stage_operations[Stage.POST_DEPLOY]
+
+
 def _get_configured_migration_stage(migration: Migration) -> Optional[Stage]:
     """Return the `Stage` configured through setting:`MIGRATION_STAGES` of the migration."""
     setting: MigrationStagesSetting = getattr(settings, "MIGRATION_STAGES", None)

tests/test_autodetector.py

@@ -80,3 +80,27 @@ def test_nullable_field_removal(self):
         changes = self.get_changes([from_model], [to_model])["tests"]
         self.assertEqual(len(changes), 1)
         self.assertEqual(get_migration_stage(changes[0]), Stage.POST_DEPLOY)
+
+    def test_mixed_stage_same_app(self):
+        from_models = [
+            ModelState(
+                "tests", "Model", [("field", models.IntegerField(primary_key=True))]
+            )
+        ]
+        to_models = [
+            ModelState(
+                "tests",
+                "OtherModel",
+                [("field", models.IntegerField(primary_key=True))],
+            )
+        ]
+        changes = self.get_changes(from_models, to_models)["tests"]
+        self.assertEqual(len(changes), 2)
+        self.assertEqual(get_migration_stage(changes[0]), Stage.PRE_DEPLOY)
+        self.assertEqual(changes[0].dependencies, [])
+        self.assertEqual(len(changes[0].operations), 1)
+        self.assertIsInstance(changes[0].operations[0], migrations.CreateModel)
+        self.assertEqual(get_migration_stage(changes[1]), Stage.POST_DEPLOY)
+        self.assertEqual(changes[1].dependencies, [("tests", "auto_1")])
+        self.assertEqual(len(changes[1].operations), 1)
+        self.assertIsInstance(changes[1].operations[0], migrations.DeleteModel)

tests/test_plan.py

@@ -5,11 +5,13 @@
 from django.test import SimpleTestCase
 
 from syzygy.constants import Stage
+from syzygy.exceptions import AmbiguousStage
 from syzygy.plan import (
     get_migration_stage,
     get_operation_stage,
     get_pre_deploy_plan,
     must_post_deploy_migration,
+    partition_operations,
 )
 
 
@@ -35,6 +37,44 @@ def test_post_deploy_operations(self):
                 self.assertIs(get_operation_stage(operation), Stage.POST_DEPLOY)
 
 
+class PartitionOperationsTests(SimpleTestCase):
+    pre_deploy_operations = [
+        CreateModel("model", []),
+    ]
+    post_deploy_operations = [
+        DeleteModel("model"),
+    ]
+
+    def test_empty(self):
+        self.assertEqual(partition_operations([]), ([], []))
+
+    def test_pre_deploy_only(self):
+        self.assertEqual(
+            partition_operations(self.pre_deploy_operations),
+            (self.pre_deploy_operations, []),
+        )
+
+    def test_post_deploy_only(self):
+        self.assertEqual(
+            partition_operations(self.post_deploy_operations),
+            ([], self.post_deploy_operations),
+        )
+
+    def test_mixed(self):
+        self.assertEqual(
+            partition_operations(
+                self.pre_deploy_operations + self.post_deploy_operations
+            ),
+            (self.pre_deploy_operations, self.post_deploy_operations),
+        )
+
+    def test_ambiguous(self):
+        with self.assertRaises(AmbiguousStage):
+            partition_operations(
+                self.post_deploy_operations + self.pre_deploy_operations
+            )
+
+
 class GetMigrationStageTests(SimpleTestCase):
     def setUp(self):
         self.migration = Migration(app_label="tests", name="migration")