Improve admin listing performance (#7318)

wagtail/admin/templatetags/wagtailadmin_tags.py

@@ -204,17 +204,17 @@ def test_page_is_public(context, page):
     Usage: {% test_page_is_public page as is_public %}
     Sets 'is_public' to True iff there are no page view restrictions in place on
     this page.
-    Caches the list of page view restrictions in the context, to avoid repeated
+    Caches the list of page view restrictions on the request, to avoid repeated
     DB queries on repeated calls.
     """
-    if 'all_page_view_restriction_paths' not in context:
-        context['all_page_view_restriction_paths'] = PageViewRestriction.objects.select_related('page').values_list(
+    if not hasattr(context["request"], "all_page_view_restriction_paths"):
+        context['request'].all_page_view_restriction_paths = PageViewRestriction.objects.select_related('page').values_list(
             'page__path', flat=True
         )
 
     is_private = any([
         page.path.startswith(restricted_path)
-        for restricted_path in context['all_page_view_restriction_paths']
+        for restricted_path in context["request"].all_page_view_restriction_paths
     ])
 
     return not is_private

wagtail/admin/views/pages/listing.py

@@ -84,6 +84,12 @@ def index(request, parent_page_id=None):
     for hook in hooks.get_hooks('construct_explorer_page_queryset'):
         pages = hook(parent_page, pages, request)
 
+    # Annotate queryset with various states to be used later for performance optimisations
+    if getattr(settings, 'WAGTAIL_WORKFLOW_ENABLED', True):
+        pages = pages.prefetch_workflow_states()
+
+    pages = pages.annotate_site_root_state().annotate_approved_schedule()
+
     # Pagination
     if do_paginate:
         paginator = Paginator(pages, per_page=50)

wagtail/core/models/__init__.py

@@ -450,7 +450,14 @@ class Page(AbstractPage, index.Indexed, ClusterableModel, metaclass=PageBase):
 
         This includes translations of site root pages as well.
         """
-        return Site.objects.filter(root_page__translation_key=self.translation_key).exists()
+        # `_is_site_root` may be populated by `annotate_site_root_state` on `PageQuerySet` as a
+        # performance optimisation
+        if hasattr(self, "_is_site_root"):
+            return self._is_site_root
+
+        return Site.objects.filter(
+            root_page__translation_key=self.translation_key
+        ).exists()
 
     @transaction.atomic
     # ensure that changes are only committed when we have updated all descendant URL paths, to preserve consistency
@@ -1451,6 +1458,11 @@ class Page(AbstractPage, index.Indexed, ClusterableModel, metaclass=PageBase):
 
     @property
     def approved_schedule(self):
+        # `_approved_schedule` may be populated by `annotate_approved_schedule` on `PageQuerySet` as a
+        # performance optimisation
+        if hasattr(self, "_approved_schedule"):
+            return self._approved_schedule
+
         return self.revisions.exclude(approved_go_live_at__isnull=True).exists()
 
     def has_unpublished_subtree(self):
@@ -2308,6 +2320,15 @@ class Page(AbstractPage, index.Indexed, ClusterableModel, metaclass=PageBase):
         """Returns True if a workflow is in progress on the current page, otherwise False"""
         if not getattr(settings, 'WAGTAIL_WORKFLOW_ENABLED', True):
             return False
+
+        # `_current_workflow_states` may be populated by `prefetch_workflow_states` on `PageQuerySet` as a
+        # performance optimisation
+        if hasattr(self, "_current_workflow_states"):
+            for state in self._current_workflow_states:
+                if state.status == WorkflowState.STATUS_IN_PROGRESS:
+                    return True
+            return False
+
         return WorkflowState.objects.filter(page=self, status=WorkflowState.STATUS_IN_PROGRESS).exists()
 
     @property
@@ -2315,6 +2336,15 @@ class Page(AbstractPage, index.Indexed, ClusterableModel, metaclass=PageBase):
         """Returns the in progress or needs changes workflow state on this page, if it exists"""
         if not getattr(settings, 'WAGTAIL_WORKFLOW_ENABLED', True):
             return None
+
+        # `_current_workflow_states` may be populated by `prefetch_workflow_states` on `pagequeryset` as a
+        # performance optimisation
+        if hasattr(self, "_current_workflow_states"):
+            try:
+                return self._current_workflow_states[0]
+            except IndexError:
+                return
+
         try:
             return WorkflowState.objects.active().select_related("current_task_state__task").get(page=self)
         except WorkflowState.DoesNotExist:

wagtail/core/query.py

@@ -5,11 +5,13 @@ from collections import defaultdict
 
 from django.apps import apps
 from django.contrib.contenttypes.models import ContentType
-from django.db.models import CharField, Q
+from django.db.models import CharField, Prefetch, Q
+from django.db.models.expressions import Exists, OuterRef
 from django.db.models.functions import Length, Substr
 from django.db.models.query import BaseIterable, ModelIterable
 from treebeard.mp_tree import MP_NodeQuerySet
 
+from wagtail.core.models.sites import Site
 from wagtail.search.queryset import SearchableQuerySetMixin
 
 
@@ -418,6 +420,57 @@ class PageQuerySet(SearchableQuerySetMixin, TreeQuerySet):
         """
         return self.exclude(self.translation_of_q(page, inclusive))
 
+    def prefetch_workflow_states(self):
+        """
+        Performance optimisation for listing pages.
+        Prefetches the active workflow states on each page in this queryset.
+        Used by `workflow_in_progress` and `current_workflow_progress` properties on
+        `wagtailcore.models.Page`.
+        """
+        from .models import WorkflowState
+
+        workflow_states = WorkflowState.objects.active().select_related(
+            "current_task_state__task"
+        )
+
+        return self.prefetch_related(
+            Prefetch(
+                "workflow_states",
+                queryset=workflow_states,
+                to_attr="_current_workflow_states",
+            )
+        )
+
+    def annotate_approved_schedule(self):
+        """
+        Performance optimisation for listing pages.
+        Annotates each page with the existence of an approved go live time.
+        Used by `approved_schedule` property on `wagtailcore.models.Page`.
+        """
+        from .models import PageRevision
+
+        return self.annotate(
+            _approved_schedule=Exists(
+                PageRevision.objects.exclude(approved_go_live_at__isnull=True).filter(
+                    page__pk=OuterRef("pk")
+                )
+            )
+        )
+
+    def annotate_site_root_state(self):
+        """
+        Performance optimisation for listing pages.
+        Annotates each object with whether it is a root page of any site.
+        Used by `is_site_root` method on `wagtailcore.models.Page`.
+        """
+        return self.annotate(
+            _is_site_root=Exists(
+                Site.objects.filter(
+                    root_page__translation_key=OuterRef("translation_key")
+                )
+            )
+        )
+
 
 def specific_iterator(qs, defer=False):
     """