Allow purging revisions of only page/non-page models

docs/reference/management_commands.md

@@ -45,13 +45,19 @@ Options:
 ## purge_revisions
 
 ```sh
-manage.py purge_revisions [--days=<number of days>]
+manage.py purge_revisions [--days=<number of days>] [--pages] [--non-pages]
 ```
 
-This command deletes old page revisions which are not in moderation, live, approved to go live, or the latest
-revision for a page. If the `days` argument is supplied, only revisions older than the specified number of
+This command deletes old revisions which are not in moderation, live, approved to go live, or the latest
+revision. If the `days` argument is supplied, only revisions older than the specified number of
 days will be deleted.
 
+If the `pages` argument is supplied, only revisions of page models will be deleted. If the `non-pages` argument is supplied, only revisions of non-page models will be deleted. If both or neither arguments are supplied, revisions of all models will be deleted.
+
+```{versionadded} 5.1
+Support for deleting revisions of non-page models is added.
+```
+
 (purge_embeds)=
 
 ## purge_embeds

wagtail/management/commands/purge_revisions.py

@@ -15,11 +15,23 @@ def add_arguments(self, parser):
             type=int,
             help="Only delete revisions older than this number of days",
         )
+        parser.add_argument(
+            "--pages",
+            action="store_true",
+            help="Only delete revisions of page models",
+        )
+        parser.add_argument(
+            "--non-pages",
+            action="store_true",
+            help="Only delete revisions of non-page models",
+        )
 
     def handle(self, *args, **options):
         days = options.get("days")
+        pages = options.get("pages")
+        non_pages = options.get("non_pages")
 
-        revisions_deleted = purge_revisions(days=days)
+        revisions_deleted = purge_revisions(days=days, pages=pages, non_pages=non_pages)
 
         if revisions_deleted:
             self.stdout.write(
@@ -31,11 +43,17 @@ def handle(self, *args, **options):
             self.stdout.write("No revisions deleted")
 
 
-def purge_revisions(days=None):
+def purge_revisions(days=None, pages=True, non_pages=True):
+    if pages == non_pages:
+        # If both are True or both are False, purge revisions of pages and non-pages
+        objects = Revision.objects.all()
+    elif pages:
+        objects = Revision.objects.page_revisions()
+    elif non_pages:
+        objects = Revision.objects.not_page_revisions()
+
     # exclude revisions which have been submitted for moderation in the old system
-    purgeable_revisions = Revision.objects.exclude(
-        submitted_for_moderation=True
-    ).exclude(
+    purgeable_revisions = objects.exclude(submitted_for_moderation=True).exclude(
         # and exclude revisions with an approved_go_live_at date
         approved_go_live_at__isnull=False
     )

wagtail/tests/test_management_commands.py

@@ -617,6 +617,8 @@ def test_future_expired_will_not_be_unpublished(self):
 
 
 class TestPurgeRevisionsCommandForPages(TestCase):
+    base_options = {}
+
     def setUp(self):
         self.object = self.get_object()
 
@@ -633,13 +635,16 @@ def get_object(self):
         self.page.refresh_from_db()
         return self.page
 
-    def run_command(self, days=None):
-        if days:
-            days_input = "--days=" + str(days)
-            return management.call_command(
-                "purge_revisions", days_input, stdout=StringIO()
-            )
-        return management.call_command("purge_revisions", stdout=StringIO())
+    def assertRevisionNotExists(self, revision):
+        self.assertFalse(Revision.objects.filter(id=revision.id).exists())
+
+    def assertRevisionExists(self, revision):
+        self.assertTrue(Revision.objects.filter(id=revision.id).exists())
+
+    def run_command(self, **options):
+        return management.call_command(
+            "purge_revisions", **{**self.base_options, **options}, stdout=StringIO()
+        )
 
     def test_latest_revision_not_purged(self):
         revision_1 = self.object.save_revision()
@@ -648,8 +653,8 @@ def test_latest_revision_not_purged(self):
         self.run_command()
 
         # revision 1 should be deleted, revision 2 should not be
-        self.assertFalse(Revision.objects.filter(id=revision_1.id).exists())
-        self.assertTrue(Revision.objects.filter(id=revision_2.id).exists())
+        self.assertRevisionNotExists(revision_1)
+        self.assertRevisionExists(revision_2)
 
     def test_revisions_in_moderation_or_workflow_not_purged(self):
         revision = self.object.save_revision(submitted_for_moderation=True)
@@ -660,7 +665,7 @@ def test_revisions_in_moderation_or_workflow_not_purged(self):
 
         self.run_command()
 
-        self.assertTrue(Revision.objects.filter(id=revision.id).exists())
+        self.assertRevisionExists(revision)
 
         workflow = Workflow.objects.create(name="test_workflow")
         task_1 = Task.objects.create(name="test_task_1")
@@ -678,13 +683,13 @@ def test_revisions_in_moderation_or_workflow_not_purged(self):
 
         # even though they're no longer the latest revisions, the old revisions
         # should stay as they are attached to an in progress workflow
-        self.assertTrue(Revision.objects.filter(id=revision.id).exists())
+        self.assertRevisionExists(revision)
 
         # If workflow is disabled at some point after that, the revision should
         # be deleted
         with override_settings(WAGTAIL_WORKFLOW_ENABLED=False):
             self.run_command()
-            self.assertFalse(Revision.objects.filter(id=revision.id).exists())
+            self.assertRevisionNotExists(revision)
 
     def test_revisions_with_approve_go_live_not_purged(self):
         revision = self.object.save_revision(
@@ -697,7 +702,7 @@ def test_revisions_with_approve_go_live_not_purged(self):
 
         self.run_command()
 
-        self.assertTrue(Revision.objects.filter(id=revision.id).exists())
+        self.assertRevisionExists(revision)
 
     def test_purge_revisions_with_date_cutoff(self):
         old_revision = self.object.save_revision()
@@ -707,22 +712,52 @@ def test_purge_revisions_with_date_cutoff(self):
         self.run_command(days=30)
 
         # revision should not be deleted, as it is younger than 30 days
-        self.assertTrue(Revision.objects.filter(id=old_revision.id).exists())
+        self.assertRevisionExists(old_revision)
 
         old_revision.created_at = timezone.now() - timedelta(days=31)
         old_revision.save()
 
         self.run_command(days=30)
 
         # revision is now older than 30 days, so should be deleted
-        self.assertFalse(Revision.objects.filter(id=old_revision.id).exists())
+        self.assertRevisionNotExists(old_revision)
 
 
 class TestPurgeRevisionsCommandForSnippets(TestPurgeRevisionsCommandForPages):
     def get_object(self):
         return FullFeaturedSnippet.objects.create(text="Hello world!")
 
 
+class TestPurgeRevisionsCommandForPagesWithPagesOnly(TestPurgeRevisionsCommandForPages):
+    base_options = {"pages": True}
+
+
+class TestPurgeRevisionsCommandForPagesWithNonPagesOnly(
+    TestPurgeRevisionsCommandForPages
+):
+    base_options = {"non_pages": True}
+
+    def assertRevisionNotExists(self, revision):
+        # Page revisions won't be purged if only non_pages is specified
+        return self.assertRevisionExists(revision)
+
+
+class TestPurgeRevisionsCommandForSnippetsWithNonPagesOnly(
+    TestPurgeRevisionsCommandForSnippets
+):
+    base_options = {"non_pages": True}
+
+
+class TestPurgeRevisionsCommandForSnippetsWithPagesOnly(
+    TestPurgeRevisionsCommandForSnippets
+):
+    base_options = {"pages": True}
+
+    def assertRevisionNotExists(self, revision):
+        # Snippet revisions won't be purged if only pages is specified
+        return self.assertRevisionExists(revision)
+
+
 class TestPurgeEmbedsCommand(TestCase):
     fixtures = ["test.json"]