added a clean command for migrations

There are situations where you may have rows in your migration table which refer to migration modules whch were deleted for some reason. The command identifies these rows, and optionally deletes them.

Author: dantownsend

piccolo/apps/migrations/commands/backwards.py

@@ -118,7 +118,7 @@ def backwards(
         Specify a value of 'all' to undo all of the migrations. Specify a
         value of '1' to undo the most recent migration.
     :param auto_agree:
-        Automatically agree to any input prompts.
+        If true, automatically agree to any input prompts.
     :param clean:
         If true, the migration files which have been run backwards are deleted
         from the disk after completing.

piccolo/apps/migrations/commands/clean.py

@@ -0,0 +1,84 @@
+from __future__ import annotations
+import typing as t
+
+from piccolo.apps.migrations.commands.base import BaseMigrationManager
+from piccolo.apps.migrations.tables import Migration
+
+
+class CleanMigrationManager(BaseMigrationManager):
+    def __init__(self, app_name: str, auto_agree: bool = False):
+        self.app_name = app_name
+        self.auto_agree = auto_agree
+        super().__init__()
+
+    def get_migration_ids_to_remove(self) -> t.List[str]:
+        """
+        Returns a list of migration ID strings, which are rows in the table,
+        but don't have a corresponding migration module on disk.
+        """
+        app_config = self.get_app_config(app_name=self.app_name)
+
+        migration_module_dict = self.get_migration_modules(
+            folder_path=app_config.migrations_folder_path
+        )
+
+        # The migration IDs which are in migration modules.
+        migration_ids = self.get_migration_ids(
+            migration_module_dict=migration_module_dict
+        )
+
+        migration_ids_to_remove = (
+            Migration.select(Migration.name)
+            .where(Migration.app_name == self.app_name)
+            .where(Migration.name.not_in(migration_ids))
+            .output(as_list=True)
+            .run_sync()
+        )
+        return migration_ids_to_remove
+
+    def run(self):
+        print("Checking the migration table ...")
+
+        # Make sure the migration table exists, otherwise we'll get an error.
+        self.create_migration_table()
+
+        migration_ids_to_remove = self.get_migration_ids_to_remove()
+
+        if migration_ids_to_remove:
+            id_string = "\n".join(migration_ids_to_remove)
+            print(
+                "The following migrations exist in the table, not not in "
+                "modules:"
+            )
+            print(id_string)
+            confirm = (
+                "y"
+                if self.auto_agree
+                else input("Would you like to delete these rows? (y/N)")
+            )
+            if confirm == "y":
+                Migration.delete().where(
+                    Migration.name.is_in(migration_ids_to_remove)
+                ).run_sync()
+                print("Deleted")
+            else:
+                print("Cancelled")
+        else:
+            print(
+                "No migrations exist in the table, which aren't also in "
+                "modules."
+            )
+
+
+def clean(app_name: str, auto_agree: bool = False):
+    """
+    Identifies any rows in the migration table which have no corresponding
+    migration module on disk, and then optionally deletes those rows.
+
+    :param app_name:
+        The name of the app to check.
+    :param auto_agree:
+        If true, automatically agree to any input prompts.
+
+    """
+    CleanMigrationManager(app_name=app_name, auto_agree=auto_agree).run()

tests/apps/migrations/commands/test_clean.py

@@ -0,0 +1,36 @@
+from unittest import TestCase
+
+from piccolo.apps.migrations.commands.clean import clean
+from piccolo.apps.migrations.tables import Migration
+
+
+class TestCleanMigrationCommand(TestCase):
+    def test_clean(self):
+        Migration.create_table(if_not_exists=True).run_sync()
+
+        real_migration_ids = [
+            "2020-12-17T18:44:30",
+            "2020-12-17T18:44:39",
+            "2020-12-17T18:44:44",
+        ]
+
+        orphaned_migration_id = "2010-01-101T00:00:00"
+
+        migration_ids = real_migration_ids + [orphaned_migration_id]
+
+        Migration.insert(
+            *[Migration(name=i, app_name="example_app") for i in migration_ids]
+        ).run_sync()
+
+        clean(app_name="example_app", auto_agree=True)
+
+        remaining_rows = (
+            Migration.select(Migration.name)
+            .where(Migration.app_name == "example_app")
+            .output(as_list=True)
+            .order_by(Migration.name)
+            .run_sync()
+        )
+        self.assertEqual(remaining_rows, real_migration_ids)
+
+        Migration.alter().drop_table(if_exists=True).run_sync()