Make migration auto detector work with Django 4.x

django/django@aa4acc1#diff-50cc08301fdd233fd64196aa90bc684d96ef557d99af67692995559646de567a

Django changd from using models to model state in the
auto detector.

Author: Photonios

psqlextra/backend/migrations/patched_autodetector.py

@@ -1,6 +1,8 @@
 from contextlib import contextmanager
 from unittest import mock
 
+import django
+
 from django.db.migrations import (
     AddField,
     AlterField,
@@ -11,7 +13,6 @@
 )
 from django.db.migrations.autodetector import MigrationAutodetector
 from django.db.migrations.operations.base import Operation
-from django.db.models import Model
 
 from psqlextra.models import (
     PostgresMaterializedViewModel,
@@ -21,6 +22,11 @@
 from psqlextra.types import PostgresPartitioningMethod
 
 from . import operations
+from .state import (
+    PostgresMaterializedViewModelState,
+    PostgresPartitionedModelState,
+    PostgresViewModelState,
+)
 
 # original `MigrationAutodetector.add_operation`
 # function, saved here so the patched version can
@@ -88,62 +94,108 @@ def _transform_view_field_operations(self, operation: Operation):
         actually applying it.
         """
 
-        model = self.autodetector.new_apps.get_model(
-            self.app_label, operation.model_name
-        )
+        if django.VERSION >= (4, 0):
+            model_identifier = (self.app_label, operation.model_name.lower())
+            model_state = (
+                self.autodetector.to_state.models.get(model_identifier)
+                or self.autodetector.from_state.models[model_identifier]
+            )
 
-        if issubclass(model, PostgresViewModel):
-            return self.add(operations.ApplyState(state_operation=operation))
+            if isinstance(model_state, PostgresViewModelState):
+                return self.add(
+                    operations.ApplyState(state_operation=operation)
+                )
+        else:
+            model = self.autodetector.new_apps.get_model(
+                self.app_label, operation.model_name
+            )
+
+            if issubclass(model, PostgresViewModel):
+                return self.add(
+                    operations.ApplyState(state_operation=operation)
+                )
 
         return self.add(operation)
 
     def add_create_model(self, operation: CreateModel):
         """Adds the specified :see:CreateModel operation to the list of
         operations to execute in the migration."""
 
-        model = self.autodetector.new_apps.get_model(
-            self.app_label, operation.name
-        )
+        if django.VERSION >= (4, 0):
+            model_state = self.autodetector.to_state.models[
+                self.app_label, operation.name.lower()
+            ]
+
+            if isinstance(model_state, PostgresPartitionedModelState):
+                return self.add_create_partitioned_model(operation)
+            elif isinstance(model_state, PostgresMaterializedViewModelState):
+                return self.add_create_materialized_view_model(operation)
+            elif isinstance(model_state, PostgresViewModelState):
+                return self.add_create_view_model(operation)
+        else:
+            model = self.autodetector.new_apps.get_model(
+                self.app_label, operation.name
+            )
 
-        if issubclass(model, PostgresPartitionedModel):
-            return self.add_create_partitioned_model(model, operation)
-        elif issubclass(model, PostgresMaterializedViewModel):
-            return self.add_create_materialized_view_model(model, operation)
-        elif issubclass(model, PostgresViewModel):
-            return self.add_create_view_model(model, operation)
+            if issubclass(model, PostgresPartitionedModel):
+                return self.add_create_partitioned_model(operation)
+            elif issubclass(model, PostgresMaterializedViewModel):
+                return self.add_create_materialized_view_model(operation)
+            elif issubclass(model, PostgresViewModel):
+                return self.add_create_view_model(operation)
 
         return self.add(operation)
 
     def add_delete_model(self, operation: DeleteModel):
         """Adds the specified :see:Deletemodel operation to the list of
         operations to execute in the migration."""
 
-        model = self.autodetector.old_apps.get_model(
-            self.app_label, operation.name
-        )
+        if django.VERSION >= (4, 0):
+            model_state = self.autodetector.from_state.models[
+                self.app_label, operation.name.lower()
+            ]
+
+            if isinstance(model_state, PostgresPartitionedModelState):
+                return self.add_delete_partitioned_model(operation)
+            elif isinstance(model_state, PostgresMaterializedViewModelState):
+                return self.add_delete_materialized_view_model(operation)
+            elif isinstance(model_state, PostgresViewModelState):
+                return self.add_delete_view_model(operation)
+        else:
+            model = self.autodetector.old_apps.get_model(
+                self.app_label, operation.name
+            )
 
-        if issubclass(model, PostgresPartitionedModel):
-            return self.add_delete_partitioned_model(model, operation)
-        elif issubclass(model, PostgresMaterializedViewModel):
-            return self.add_delete_materialized_view_model(model, operation)
-        elif issubclass(model, PostgresViewModel):
-            return self.add_delete_view_model(model, operation)
+            if issubclass(model, PostgresPartitionedModel):
+                return self.add_delete_partitioned_model(operation)
+            elif issubclass(model, PostgresMaterializedViewModel):
+                return self.add_delete_materialized_view_model(operation)
+            elif issubclass(model, PostgresViewModel):
+                return self.add_delete_view_model(operation)
 
         return self.add(operation)
 
-    def add_create_partitioned_model(
-        self, model: Model, operation: CreateModel
-    ):
+    def add_create_partitioned_model(self, operation: CreateModel):
         """Adds a :see:PostgresCreatePartitionedModel operation to the list of
         operations to execute in the migration."""
 
-        partitioning_options = model._partitioning_meta.original_attrs
+        if django.VERSION >= (4, 0):
+            model_state = self.autodetector.to_state.models[
+                self.app_label, operation.name.lower()
+            ]
+            partitioning_options = model_state.partitioning_options
+        else:
+            model = self.autodetector.new_apps.get_model(
+                self.app_label, operation.name
+            )
+            partitioning_options = model._partitioning_meta.original_attrs
+
         _, args, kwargs = operation.deconstruct()
 
         if partitioning_options["method"] != PostgresPartitioningMethod.HASH:
             self.add(
                 operations.PostgresAddDefaultPartition(
-                    model_name=model.__name__, name="default"
+                    model_name=operation.name, name="default"
                 )
             )
 
@@ -153,9 +205,7 @@ def add_create_partitioned_model(
             )
         )
 
-    def add_delete_partitioned_model(
-        self, model: Model, operation: DeleteModel
-    ):
+    def add_delete_partitioned_model(self, operation: DeleteModel):
         """Adds a :see:PostgresDeletePartitionedModel operation to the list of
         operations to execute in the migration."""
 
@@ -164,11 +214,21 @@ def add_delete_partitioned_model(
             operations.PostgresDeletePartitionedModel(*args, **kwargs)
         )
 
-    def add_create_view_model(self, model: Model, operation: CreateModel):
+    def add_create_view_model(self, operation: CreateModel):
         """Adds a :see:PostgresCreateViewModel operation to the list of
         operations to execute in the migration."""
 
-        view_options = model._view_meta.original_attrs
+        if django.VERSION >= (4, 0):
+            model_state = self.autodetector.to_state.models[
+                self.app_label, operation.name.lower()
+            ]
+            view_options = model_state.view_options
+        else:
+            model = self.autodetector.new_apps.get_model(
+                self.app_label, operation.name
+            )
+            view_options = model._view_meta.original_attrs
+
         _, args, kwargs = operation.deconstruct()
 
         self.add(
@@ -177,20 +237,28 @@ def add_create_view_model(self, model: Model, operation: CreateModel):
             )
         )
 
-    def add_delete_view_model(self, model: Model, operation: DeleteModel):
+    def add_delete_view_model(self, operation: DeleteModel):
         """Adds a :see:PostgresDeleteViewModel operation to the list of
         operations to execute in the migration."""
 
         _, args, kwargs = operation.deconstruct()
         return self.add(operations.PostgresDeleteViewModel(*args, **kwargs))
 
-    def add_create_materialized_view_model(
-        self, model: Model, operation: CreateModel
-    ):
+    def add_create_materialized_view_model(self, operation: CreateModel):
         """Adds a :see:PostgresCreateMaterializedViewModel operation to the
         list of operations to execute in the migration."""
 
-        view_options = model._view_meta.original_attrs
+        if django.VERSION >= (4, 0):
+            model_state = self.autodetector.to_state.models[
+                self.app_label, operation.name.lower()
+            ]
+            view_options = model_state.view_options
+        else:
+            model = self.autodetector.new_apps.get_model(
+                self.app_label, operation.name
+            )
+            view_options = model._view_meta.original_attrs
+
         _, args, kwargs = operation.deconstruct()
 
         self.add(
@@ -199,9 +267,7 @@ def add_create_materialized_view_model(
             )
         )
 
-    def add_delete_materialized_view_model(
-        self, model: Model, operation: DeleteModel
-    ):
+    def add_delete_materialized_view_model(self, operation: DeleteModel):
         """Adds a :see:PostgresDeleteMaterializedViewModel operation to the
         list of operations to execute in the migration."""