Update Page.objects.specific() to warn instead of erroring when pages of the specific type cannot be found (#5928)

CHANGELOG.txt

@@ -45,6 +45,7 @@ Changelog
  * Disable password auto-completion on user creation form (Samir Shah)
  * Upgrade jQuery to version 3.5.1 to reduce penetration testing false positives (Matt Westcott)
  * Add ability to extend `EditHandler` without a children attribute (Seb Brown)
+ * `Page.objects.specific` now gracefully handles pages with missing specific records (Andy Babic)
  * Fix: Support IPv6 domain (Alex Gleason, Coen van der Kamp)
  * Fix: Ensure link to add a new user works when no users are visible in the users list (LB (Ben Johnston))
  * Fix: `AbstractEmailForm` saved submission fields are now aligned with the email content fields, `form.cleaned_data` will be used instead of `form.fields` (Haydn Greatnews)

docs/releases/2.10.rst

@@ -63,6 +63,7 @@ Other features
  * Disable password auto-completion on user creation form (Samir Shah)
  * Upgrade jQuery to version 3.5.1 to reduce penetration testing false positives (Matt Westcott)
  * Add ability to extend ``EditHandler`` without a children attribute (Seb Brown)
+ * ``Page.objects.specific`` now gracefully handles pages with missing specific records (Andy Babic)
 
 
 Bug fixes

wagtail/core/query.py

@@ -1,4 +1,5 @@
 import posixpath
+import warnings
 from collections import defaultdict
 
 from django.apps import apps
@@ -373,6 +374,8 @@ def specific_iterator(qs, defer=False):
 
     This should be called from ``PageQuerySet.specific``
     """
+    from wagtail.core.models import Page
+
     annotation_aliases = qs.query.annotations.keys()
     values = qs.values('pk', 'content_type', *annotation_aliases)
 
@@ -393,29 +396,50 @@ def specific_iterator(qs, defer=False):
 
     # Get the specific instances of all pages, one model class at a time.
     pages_by_type = {}
+    missing_pks = []
+
     for content_type, pks in pks_by_type.items():
         # look up model class for this content type, falling back on the original
         # model (i.e. Page) if the more specific one is missing
         model = content_types[content_type].model_class() or qs.model
         pages = model.objects.filter(pk__in=pks)
 
-        if annotation_aliases:
-            # Reapply annotations to pages.
-            for page in pages:
-                for annotation, value in annotations_by_pk.get(page.pk, {}).items():
-                    setattr(page, annotation, value)
-
         if defer:
             # Defer all specific fields
-            from wagtail.core.models import Page
             fields = [field.attname for field in Page._meta.get_fields() if field.concrete]
             pages = pages.only(*fields)
 
-        pages_by_type[content_type] = {page.pk: page for page in pages}
+        pages_for_type = {page.pk: page for page in pages}
+        pages_by_type[content_type] = pages_for_type
+        missing_pks.extend(
+            pk for pk in pks if pk not in pages_for_type
+        )
 
-    # Yield all of the pages, in the order they occurred in the original query.
+    # Fetch generic pages to supplement missing items
+    if missing_pks:
+        generic_pages = Page.objects.filter(pk__in=missing_pks).select_related('content_type').in_bulk()
+        warnings.warn(
+            "Specific versions of the following pages could not be found. "
+            "This is most likely because a database migration has removed "
+            "the relevant table or record since the page was created:\n{}".format([
+                {'id': p.id, 'title': p.title, 'type': p.content_type}
+                for p in generic_pages.values()
+            ]), category=RuntimeWarning
+        )
+    else:
+        generic_pages = {}
+
+    # Yield all pages in the order they occurred in the original query.
     for pk, content_type in pks_and_types:
-        yield pages_by_type[content_type][pk]
+        try:
+            page = pages_by_type[content_type][pk]
+        except KeyError:
+            page = generic_pages[pk]
+        if annotation_aliases:
+            # Reapply annotations before returning
+            for annotation, value in annotations_by_pk.get(page.pk, {}).items():
+                setattr(page, annotation, value)
+        yield page
 
 
 class SpecificIterable(BaseIterable):

wagtail/core/tests/test_page_queryset.py

@@ -1,3 +1,5 @@
+from unittest import mock
+
 from django.contrib.contenttypes.models import ContentType
 from django.db.models import Count, Q
 from django.test import TestCase
@@ -670,6 +672,25 @@ class TestSpecificQuery(TestCase):
             Page.objects.get(url_path='/home/other/').specific,
         ])
 
+    def test_specific_gracefully_handles_missing_rows(self):
+        # 5928 - PageQuerySet.specific should gracefully handle pages whose ContentType
+        # row in the specific table no longer exists
+
+        # Trick specific_iterator into always looking for EventPages
+        with mock.patch(
+            'wagtail.core.query.ContentType.objects.get_for_id',
+            return_value=ContentType.objects.get_for_model(EventPage),
+        ):
+            with self.assertWarnsRegex(RuntimeWarning, "Specific versions of the following pages could not be found"):
+                pages = list(Page.objects.get(url_path='/home/').get_children().specific())
+
+            # All missing pages should be supplemented with generic pages
+            self.assertEqual(pages, [
+                Page.objects.get(url_path='/home/events/'),
+                Page.objects.get(url_path='/home/about-us/'),
+                Page.objects.get(url_path='/home/other/'),
+            ])
+
     def test_deferred_specific_query(self):
         # Tests the "defer" keyword argument, which defers all specific fields
         root = Page.objects.get(url_path='/home/')