Implement aliased model rename.

syzygy/operations.py

@@ -1,5 +1,6 @@
 from contextlib import contextmanager
 
+from django.db import transaction
 from django.db.migrations import operations
 from django.db.models.fields import NOT_PROVIDED
 
@@ -255,3 +256,95 @@ class AlterField(StagedOperation, operations.AlterField):
     """
     Subclass of ``AlterField`` that allows explicitly defining a stage.
     """
+
+
+def _create_instead_of_triggers(schema_editor, old_model, new_model):
+    quote = schema_editor.quote_name
+    schema_editor.execute(
+        (
+            "CREATE TRIGGER {trigger_name} INSTEAD OF INSERT ON {old_table}\n"
+            "BEGIN\n"
+            "INSERT INTO {new_table}({fields}) VALUES({values});\n"
+            "END"
+        ).format(
+            trigger_name=f"{old_model._meta.db_table}_insert",
+            old_table=quote(old_model._meta.db_table),
+            new_table=quote(new_model._meta.db_table),
+            fields=", ".join(
+                quote(field.column) for field in old_model._meta.local_fields
+            ),
+            values=", ".join(
+                f"NEW.{quote(field.column)}" for field in old_model._meta.local_fields
+            ),
+        )
+    )
+    for field in old_model._meta.local_fields:
+        schema_editor.execute(
+            (
+                "CREATE TRIGGER {trigger_name} INSTEAD OF UPDATE OF {column} ON {old_table}\n"
+                "BEGIN\n"
+                "UPDATE {new_table} SET {column}=NEW.{column} WHERE {pk}=NEW.{pk};\n"
+                "END"
+            ).format(
+                trigger_name=f"{old_model._meta.db_table}_update_{field.column}",
+                old_table=quote(old_model._meta.db_table),
+                new_table=quote(new_model._meta.db_table),
+                column=quote(field.column),
+                pk=quote(new_model._meta.pk.column),
+            )
+        )
+    schema_editor.execute(
+        (
+            "CREATE TRIGGER {trigger_name} INSTEAD OF DELETE ON {old_table}\n"
+            "BEGIN\n"
+            "DELETE FROM {new_table} WHERE {pk}=OLD.{pk};\n"
+            "END"
+        ).format(
+            trigger_name=f"{old_model._meta.db_table}_delete",
+            old_table=quote(old_model._meta.db_table),
+            new_table=quote(new_model._meta.db_table),
+            pk=quote(new_model._meta.pk.column),
+        )
+    )
+
+
+class AliasedRenameModel(operations.RenameModel):
+    stage = Stage.PRE_DEPLOY
+
+    def database_forwards(self, app_label, schema_editor, from_state, to_state):
+        new_model = to_state.apps.get_model(app_label, self.new_name)
+        alias = schema_editor.connection.alias
+        if not self.allow_migrate_model(alias, new_model):
+            return
+        old_model = from_state.apps.get_model(app_label, self.old_name)
+        quote = schema_editor.quote_name
+        with transaction.atomic(alias):
+            super().database_forwards(app_label, schema_editor, from_state, to_state)
+            schema_editor.execute(
+                "CREATE VIEW {} AS SELECT * FROM {}".format(
+                    quote(old_model._meta.db_table), quote(new_model._meta.db_table)
+                )
+            )
+            if schema_editor.connection.vendor == "sqlite":
+                _create_instead_of_triggers(schema_editor, old_model, new_model)
+
+    def database_backwards(self, app_label, schema_editor, from_state, to_state):
+        new_model = to_state.apps.get_model(app_label, self.new_name)
+        alias = schema_editor.connection.alias
+        if not self.allow_migrate_model(alias, new_model):
+            return
+        old_model = from_state.apps.get_model(app_label, self.old_name)
+        with transaction.atomic(alias):
+            schema_editor.execute(
+                "DROP VIEW {}".format(
+                    schema_editor.quote_name(old_model._meta.db_table)
+                )
+            )
+            super().database_backwards(app_label, schema_editor, from_state, to_state)
+
+    def describe(self):
+        return "Rename model %s to %s while creating an alias for %s" % (
+            self.old_name,
+            self.new_name,
+            self.old_name,
+        )

tests/test_operations.py

@@ -10,7 +10,12 @@
 
 from syzygy.autodetector import MigrationAutodetector
 from syzygy.constants import Stage
-from syzygy.operations import AddField, PostAddField, PreRemoveField
+from syzygy.operations import (
+    AddField,
+    AliasedRenameModel,
+    PostAddField,
+    PreRemoveField,
+)
 from syzygy.plan import get_operation_stage
 
 
@@ -313,3 +318,27 @@ def test_elidable(self):
             migrations.RemoveField(model_name, field_name, field),
         ]
         self.assert_optimizes_to(operations, [operations[-1]])
+
+
+class AliasedRenameModelTests(OperationTestCase):
+    def test_database_forwards(self):
+        model_name = "TestModel"
+        field = models.IntegerField()
+        state = self.apply_operations([
+            migrations.CreateModel(model_name, [("foo", field)]),
+        ])
+        pre_model = state.apps.get_model("tests", model_name)
+        new_model_name = "NewTestModel"
+        state = self.apply_operations([
+            AliasedRenameModel(model_name, new_model_name),
+        ], state)
+        pre_obj = pre_model.objects.create(foo=1)
+        self.assertEqual(pre_model.objects.get(), pre_obj)
+        post_model = state.apps.get_model("tests", new_model_name)
+        self.assertEqual(post_model.objects.get().pk, pre_obj.pk)
+        pre_model.objects.all().delete()
+        post_obj = post_model.objects.create(foo=2)
+        self.assertEqual(post_model.objects.get(), post_obj)
+        self.assertEqual(pre_model.objects.get().pk, post_obj.pk)
+        pre_model.objects.update(foo=3)
+        self.assertEqual(post_model.objects.get().foo, 3)