Add check for separate runpython data migration

Vibed with ClaudeCode

Author: SleipRecx

docs/incompatibilities.md

@@ -38,6 +38,7 @@ You can ignore checks through the `--exclude-migration-tests` option by specifyi
 | `RUNPYTHON_ARGS_NAMING_CONVENTION` | By convention, RunPython names two arguments: apps, schema_editor                                                                                                                                                                                                                                                                 | Warning      |
 | `RUNPYTHON_MODEL_IMPORT`           | Missing apps.get_model() calls for model                                                                                                                                                                                                                                                                                          | Error        |
 | `RUNPYTHON_MODEL_VARIABLE_NAME`    | The model variable name is different from the model class itself                                                                                                                                                                                                                                                                  | Warning      |
+| `RUNPYTHON_MIXED_OPERATIONS`       | RunPython operation is mixed with schema operations (data migrations should be in separate files)                                                                                                                                                                                                                                 | Error        |
 | `RUNSQL_REVERSIBLE`                | RunSQL data migration is not reversible (missing reverse SQL)                                                                                                                                                                                                                                                                     | Warning      |
 | `CREATE_INDEX`                     | (Postgresql specific) Creating an index without the concurrent keyword will lock the table and may generate downtime                                                                                                                                                                                                              | Warning      |
 | `CREATE_INDEX_EXCLUSIVE`           | (Postgresql specific) Creating an index in a transaction acquiring an `EXCLUSIVE` lock (e.g. most `ALTER TABLE` statements acquire one) prolongs the exclusive lock on the table. Using concurrently clause does not address the issue. On the contrary, it prolongs the transaction, making it more dangerous in this situation. | Warning      |
@@ -177,6 +178,51 @@ It could happen that you use a `RunPython` operation to fill a new column. But i
 
 :white_check_mark: **Solution**: use `apps.get_model` to get the model class
 
+### :arrow_forward: Mixing RunPython with schema operations
+
+Mixing RunPython data migrations with schema operations (like AddField, AlterField, etc.) in the same migration file can lead to serious production issues.
+
+**Why this is problematic**:
+
+1. **Database lock issues**: When running RunPython with data chunking in a transactional migration, you may hold database locks for extended periods, potentially causing downtime
+2. **Atomic flag complications**: If you set `atomic = False` on a migration that contains both data and schema operations, schema changes won't rollback if the data migration fails
+3. **Rollback problems**: When mixing operations, failures in data migration can leave schema changes applied but data changes reverted, creating inconsistent database states
+
+**Forward migration**:
+1. Migration adds a new field and runs RunPython to populate it
+2. If the RunPython operation fails mid-way through a large dataset, the schema change might be committed but data is only partially migrated
+3. Your application may crash because it expects the field to be fully populated
+
+**Rollback**:
+1. If you set `atomic = False` to prevent locks, rolling back becomes problematic
+2. Schema changes may have been committed separately from data changes
+3. You lose the safety of transactional rollback
+
+:white_check_mark: **Solution**: Always separate data migrations into their own files
+
+**Good pattern** (three separate files):
+```python
+# 0001_add_slug_field.py
+operations = [
+    migrations.AddField('model', 'slug', models.CharField(default='')),
+]
+
+# 0002_backfill_slug.py
+class Migration(migrations.Migration):
+    atomic = False  # Safe to set now, only data operations
+
+    operations = [
+        migrations.RunPython(backfill_slug_values),
+    ]
+
+# 0003_make_slug_required.py
+operations = [
+    migrations.AlterField('model', 'slug', models.CharField()),
+]
+```
+
+**Note**: Combining RunPython with RunSQL is acceptable, as both are data migration operations.
+
 ## The special case of sqlite
 
 While on PostgreSQL and MySQL a table modification can be expressed by one `ALTER TABLE` statement, sqlite is handled in a different way.

src/django_migration_linter/migration_linter.py

@@ -415,7 +415,8 @@ def _gather_migrations_git(
 
         migrations = []
         for line in map(
-            clean_bytes_to_str, diff_process.stdout.readlines()  # type: ignore
+            clean_bytes_to_str,
+            diff_process.stdout.readlines(),  # type: ignore
         ):
             # Only gather lines that include added migrations.
             if self.is_migration_file(line):
@@ -455,7 +456,8 @@ def _gather_migrations_git(
         if diff_process.returncode != 0:
             output = []
             for line in map(
-                clean_bytes_to_str, diff_process.stderr.readlines()  # type: ignore
+                clean_bytes_to_str,
+                diff_process.stderr.readlines(),  # type: ignore
             ):
                 output.append(line)
             logger.error("Error while git diff command:\n{}".format("".join(output)))
@@ -529,6 +531,17 @@ def analyse_data_migration(
             if op_warnings:
                 warnings += op_warnings
 
+        # Check for mixed RunPython and schema operations
+        mixed_errors, mixed_ignored, mixed_warnings = (
+            self.check_mixed_runpython_operations(migration.operations)
+        )
+        if mixed_errors:
+            errors += mixed_errors
+        if mixed_ignored:
+            ignored += mixed_ignored
+        if mixed_warnings:
+            warnings += mixed_warnings
+
         return errors, ignored, warnings
 
     @staticmethod
@@ -764,3 +777,63 @@ def lint_runsql(
                 warning += sql_warnings
 
         return error, ignored, warning
+
+    def check_mixed_runpython_operations(
+        self, operations: Iterable[Operation]
+    ) -> tuple[list[Issue], list[Issue], list[Issue]]:
+        """
+        Check if migration mixes RunPython with schema operations.
+
+        Data migrations (RunPython/RunSQL) should be in separate migration files
+        from schema operations to avoid database lock issues and ensure proper
+        rollback behavior.
+
+        Returns:
+            Tuple of (errors, ignored, warnings) where each is a list of Issue objects
+        """
+        errors: list[Issue] = []
+        ignored: list[Issue] = []
+        warnings: list[Issue] = []
+
+        # Check if there are any RunPython operations
+        has_runpython = any(isinstance(op, RunPython) for op in operations)
+
+        if not has_runpython:
+            # No RunPython, nothing to check
+            return errors, ignored, warnings
+
+        # Check for other operation types (schema operations)
+        # RunSQL is allowed with RunPython as both are data migrations
+        # IgnoreMigration is a marker operation
+        # SeparateDatabaseAndState is OK with RunPython (advanced usage)
+        schema_operations = [
+            op
+            for op in operations
+            if not isinstance(op, (RunPython, RunSQL, IgnoreMigration))
+            and op.__class__.__name__ != "SeparateDatabaseAndState"
+        ]
+
+        if schema_operations:
+            # Get unique operation names for better error message
+            operation_names = list(
+                {op.__class__.__name__ for op in schema_operations[:3]}
+            )
+            operation_str = ", ".join(operation_names)
+            if len(schema_operations) > 3:
+                operation_str += "..."
+
+            issue = Issue(
+                code="RUNPYTHON_MIXED_OPERATIONS",
+                message=(
+                    "RunPython operation is mixed with schema operations. "
+                    "Data migrations should be in separate files. "
+                    f"Found schema operations: {operation_str}"
+                ),
+            )
+
+            if issue.code in self.exclude_migration_tests:
+                ignored.append(issue)
+            else:
+                errors.append(issue)
+
+        return errors, ignored, warnings

tests/fixtures.py

@@ -16,6 +16,7 @@
 DROP_UNIQUE_TOGETHER = "app_unique_together"
 ADD_MANYTOMANY_FIELD = "app_add_manytomany_field"
 DATA_MIGRATIONS = "app_data_migrations"
+MIXED_RUNPYTHON_OPERATIONS = "app_mixed_runpython_operations"
 MAKE_NOT_NULL_WITH_DJANGO_DEFAULT = "app_make_not_null_with_django_default"
 MAKE_NOT_NULL_WITH_LIB_DEFAULT = "app_make_not_null_with_lib_default"
 CREATE_INDEX_EXCLUSIVE = "app_create_index_exclusive"

tests/functional/test_data_migrations.py

@@ -306,6 +306,67 @@ def forward_op(apps, schema_editor):
         self.assertEqual(1, len(issues))
 
 
+class MixedRunPythonOperationsTestCase(unittest.TestCase):
+    def setUp(self):
+        test_project_path = os.path.dirname(settings.BASE_DIR)
+        self.linter = MigrationLinter(
+            test_project_path,
+            include_apps=fixtures.MIXED_RUNPYTHON_OPERATIONS,
+        )
+
+    def test_mixed_runpython_with_schema_operations(self):
+        """Test that mixing RunPython with schema operations produces an error."""
+        mixed_migration = self.linter.migration_loader.disk_migrations[
+            ("app_mixed_runpython_operations", "0002_mixed_operations")
+        ]
+        self.linter.lint_migration(mixed_migration)
+
+        self.assertEqual(1, self.linter.nb_erroneous)
+        self.assertTrue(self.linter.has_errors)
+        self.assertEqual(0, self.linter.nb_valid)
+
+    def test_runpython_only(self):
+        """Test that RunPython-only migrations pass without errors."""
+        runpython_only_migration = self.linter.migration_loader.disk_migrations[
+            ("app_mixed_runpython_operations", "0003_runpython_only")
+        ]
+        self.linter.lint_migration(runpython_only_migration)
+
+        self.assertEqual(0, self.linter.nb_erroneous)
+        self.assertFalse(self.linter.has_errors)
+        # Might have warnings for other checks (reversible, etc), but no errors
+        self.assertTrue(self.linter.nb_valid >= 1 or self.linter.nb_warnings >= 1)
+
+    def test_runpython_with_runsql(self):
+        """Test that RunPython with RunSQL is allowed (both are data migrations)."""
+        mixed_data_migration = self.linter.migration_loader.disk_migrations[
+            ("app_mixed_runpython_operations", "0004_runpython_with_runsql")
+        ]
+        self.linter.lint_migration(mixed_data_migration)
+
+        self.assertEqual(0, self.linter.nb_erroneous)
+        self.assertFalse(self.linter.has_errors)
+
+    def test_exclude_mixed_operations_check(self):
+        """Test that RUNPYTHON_MIXED_OPERATIONS can be excluded."""
+        self.linter = MigrationLinter(
+            os.path.dirname(settings.BASE_DIR),
+            include_apps=fixtures.MIXED_RUNPYTHON_OPERATIONS,
+            exclude_migration_tests=("RUNPYTHON_MIXED_OPERATIONS",),
+        )
+
+        mixed_migration = self.linter.migration_loader.disk_migrations[
+            ("app_mixed_runpython_operations", "0002_mixed_operations")
+        ]
+        self.linter.lint_migration(mixed_migration)
+
+        # Should not error when excluded
+        self.assertEqual(0, self.linter.nb_erroneous)
+        self.assertFalse(self.linter.has_errors)
+        # Should be valid since the check is excluded
+        self.assertEqual(1, self.linter.nb_valid)
+
+
 class RunSQLMigrationTestCase(unittest.TestCase):
     def setUp(self):
         test_project_path = os.path.dirname(settings.BASE_DIR)

tests/test_project/app_mixed_runpython_operations/__init__.py

@@ -0,0 +1,29 @@
+# Generated by Django 5.0 on 2025-10-22 11:00
+
+from __future__ import annotations
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    initial = True
+
+    dependencies = []
+
+    operations = [
+        migrations.CreateModel(
+            name="TestModel",
+            fields=[
+                (
+                    "id",
+                    models.AutoField(
+                        auto_created=True,
+                        primary_key=True,
+                        serialize=False,
+                        verbose_name="ID",
+                    ),
+                ),
+                ("name", models.CharField(max_length=100)),
+            ],
+        )
+    ]

tests/test_project/app_mixed_runpython_operations/migrations/0001_initial.py

@@ -0,0 +1,35 @@
+# Generated by Django 5.0 on 2025-10-22 11:01
+
+from __future__ import annotations
+
+from django.db import migrations, models
+
+
+def set_slug_from_name(apps, schema_editor):
+    TestModel = apps.get_model("app_mixed_runpython_operations", "TestModel")
+    for obj in TestModel.objects.all():
+        obj.slug = obj.name.lower().replace(" ", "-")
+        obj.save()
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("app_mixed_runpython_operations", "0001_initial"),
+    ]
+
+    operations = [
+        # This is BAD - mixing schema operations with RunPython
+        migrations.AddField(
+            model_name="testmodel",
+            name="slug",
+            field=models.CharField(max_length=100, default="", null=True),
+        ),
+        migrations.RunPython(
+            set_slug_from_name, reverse_code=migrations.RunPython.noop
+        ),
+        migrations.AlterField(
+            model_name="testmodel",
+            name="slug",
+            field=models.CharField(max_length=100, null=True),
+        ),
+    ]

tests/test_project/app_mixed_runpython_operations/migrations/0002_mixed_operations.py

@@ -0,0 +1,23 @@
+# Generated by Django 5.0 on 2025-10-22 11:02
+
+from __future__ import annotations
+
+from django.db import migrations
+
+
+def backfill_data(apps, schema_editor):
+    TestModel = apps.get_model("app_mixed_runpython_operations", "TestModel")
+    for obj in TestModel.objects.filter(slug__isnull=True):
+        obj.slug = "default"
+        obj.save()
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("app_mixed_runpython_operations", "0002_mixed_operations"),
+    ]
+
+    operations = [
+        # This is GOOD - only RunPython operations
+        migrations.RunPython(backfill_data, reverse_code=migrations.RunPython.noop),
+    ]

tests/test_project/app_mixed_runpython_operations/migrations/0003_runpython_only.py

@@ -0,0 +1,27 @@
+# Generated by Django 5.0 on 2025-10-22 11:03
+
+from __future__ import annotations
+
+from django.db import migrations
+
+
+def update_data(apps, schema_editor):
+    TestModel = apps.get_model("app_mixed_runpython_operations", "TestModel")
+    for obj in TestModel.objects.all():
+        obj.slug = obj.slug.upper()
+        obj.save()
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("app_mixed_runpython_operations", "0003_runpython_only"),
+    ]
+
+    operations = [
+        # This is GOOD - RunPython with RunSQL is fine (both are data migrations)
+        migrations.RunPython(update_data, reverse_code=migrations.RunPython.noop),
+        migrations.RunSQL(
+            "UPDATE app_mixed_runpython_operations_testmodel SET name = UPPER(name);",
+            reverse_sql=migrations.RunSQL.noop,
+        ),
+    ]