Refactor page fetching logic to cache per request (#11683)

Adds two new helper static methods:
- `Page.route_for_request()` - to find the page route, given a request
  object and a URL path
- `Page.find_for_request()` - to find the page given, a request object and a URL
  path

docs/reference/pages/model_reference.md

@@ -186,6 +186,10 @@ See also [django-treebeard](https://django-treebeard.readthedocs.io/en/latest/in
 
     .. automethod:: serve
     
+    .. automethod:: route_for_request
+    
+    .. automethod:: find_for_request
+
     .. autoattribute:: context_object_name
 
         Custom name for page instance in page's ``Context``.

wagtail/models/__init__.py

@@ -9,11 +9,14 @@ should implement low-level generic functionality which is then imported by highe
 as Page.
 """
 
+from __future__ import annotations
+
 import functools
 import logging
 import posixpath
 import uuid
 from io import StringIO
+from typing import TYPE_CHECKING
 from urllib.parse import urlparse
 
 from django import forms
@@ -126,6 +129,9 @@ from .sites import Site, SiteManager, SiteRootPath  # noqa: F401
 from .specific import SpecificMixin
 from .view_restrictions import BaseViewRestriction
 
+if TYPE_CHECKING:
+    from django.http import HttpRequest
+
 logger = logging.getLogger("wagtail")
 
 PAGE_TEMPLATE_VAR = "page"
@@ -1283,6 +1289,43 @@ class Page(AbstractPage, index.Indexed, ClusterableModel, metaclass=PageBase):
     promote_panels = []
     settings_panels = []
 
+    @staticmethod
+    def route_for_request(request: "HttpRequest", path: str) -> RouteResult | None:
+        """
+        Find the page route for the given HTTP request object, and URL path. The route
+        result (`page`, `args`, and `kwargs`) will be cached via
+        `request._wagtail_route_for_request`.
+        """
+        if not hasattr(request, "_wagtail_route_for_request"):
+            try:
+                # we need a valid Site object for this request in order to proceed
+                if site := Site.find_for_request(request):
+                    path_components = [
+                        component for component in path.split("/") if component
+                    ]
+                    request._wagtail_route_for_request = (
+                        site.root_page.localized.specific.route(
+                            request, path_components
+                        )
+                    )
+                else:
+                    request._wagtail_route_for_request = None
+            except Http404:
+                # .route() can raise Http404
+                request._wagtail_route_for_request = None
+
+        return request._wagtail_route_for_request
+
+    @staticmethod
+    def find_for_request(request: "HttpRequest", path: str) -> "Page" | None:
+        """
+        Find the page for the given HTTP request object, and URL path. The full
+        page route will be cached via `request._wagtail_route_for_request`
+        """
+        result = Page.route_for_request(request, path)
+        if result is not None:
+            return result[0]
+
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
         if not self.id:

wagtail/test/middleware.py

@@ -1,6 +1,9 @@
-from django.http import HttpResponseForbidden
+from django.http import Http404, HttpResponse, HttpResponseForbidden
 from django.utils.deprecation import MiddlewareMixin
 
+from wagtail.models import Page
+from wagtail.views import serve
+
 
 class BlockDodgyUserAgentMiddleware(MiddlewareMixin):
     # Used to test that we're correctly handling responses returned from middleware during page
@@ -16,3 +19,13 @@ class BlockDodgyUserAgentMiddleware(MiddlewareMixin):
             and request.headers.get("user-agent") == "EvilHacker"
         ):
             return HttpResponseForbidden("Forbidden")
+
+
+class SimplePageViewInterceptorMiddleware(MiddlewareMixin):
+    def process_view(self, request, view_func, view_args, view_kwargs):
+        if serve == view_func:
+            page = Page.find_for_request(request, *view_args, **view_kwargs)
+            if page is None:
+                raise Http404
+            elif page.content == "Intercept me":
+                return HttpResponse("Intercepted")

wagtail/tests/test_page_model.py

@@ -68,6 +68,7 @@ from wagtail.test.testapp.models import (
     TaggedPage,
 )
 from wagtail.test.utils import WagtailTestUtils
+from wagtail.url_routing import RouteResult
 
 
 def get_ct(model):
@@ -211,6 +212,50 @@ class TestSiteRouting(TestCase):
         self.unrecognised_port = "8000"
         self.unrecognised_hostname = "unknown.site.com"
 
+    def test_route_for_request_query_count(self):
+        request = get_dummy_request(site=self.events_site)
+        with self.assertNumQueries(2):
+            # expect queries for site & page
+            Page.route_for_request(request, request.path)
+        with self.assertNumQueries(0):
+            # subsequent lookups should be cached on the request
+            Page.route_for_request(request, request.path)
+
+    def test_route_for_request_value(self):
+        request = get_dummy_request(site=self.events_site)
+        self.assertFalse(hasattr(request, "_wagtail_route_for_request"))
+        result = Page.route_for_request(request, request.path)
+        self.assertTrue(isinstance(result, RouteResult))
+        self.assertEqual(
+            (result[0], result[1], result[2]),
+            (self.events_site.root_page.specific, [], {}),
+        )
+        self.assertTrue(hasattr(request, "_wagtail_route_for_request"))
+        self.assertIs(request._wagtail_route_for_request, result)
+
+    def test_route_for_request_cached(self):
+        request = get_dummy_request(site=self.events_site)
+        m = Mock()
+        request._wagtail_route_for_request = m
+        with self.assertNumQueries(0):
+            self.assertEqual(Page.route_for_request(request, request.path), m)
+
+    def test_route_for_request_suppresses_404(self):
+        request = get_dummy_request(path="does-not-exist", site=self.events_site)
+        self.assertIsNone(Page.route_for_request(request, request.path))
+
+    def test_find_for_request(self):
+        request_200 = get_dummy_request(site=self.events_site)
+        self.assertEqual(
+            Page.find_for_request(request_200, request_200.path),
+            self.events_site.root_page.specific,
+        )
+        request_404 = get_dummy_request(path="does-not-exist", site=self.events_site)
+        self.assertEqual(
+            Page.find_for_request(request_404, request_404.path),
+            None,
+        )
+
     def test_valid_headers_route_to_specific_site(self):
         # requests with a known Host: header should be directed to the specific site
         request = get_dummy_request(site=self.events_site)

wagtail/tests/test_views.py

@@ -1,8 +1,13 @@
+from unittest import mock
+
 from django.test import TestCase
 from django.urls import reverse
 
-from wagtail.models import Page
+from wagtail.coreutils import get_dummy_request
+from wagtail.models import Page, Site
+from wagtail.test.testapp.models import SimplePage
 from wagtail.test.utils import WagtailTestUtils
+from wagtail.views import serve
 
 
 class TestLoginView(WagtailTestUtils, TestCase):
@@ -47,3 +52,49 @@ class TestLoginView(WagtailTestUtils, TestCase):
             },
         )
         self.assertRedirects(response, self.events_index.url)
+
+
+@mock.patch("wagtail.hooks.get_hooks", mock.Mock(return_value=[]))
+class TestServeView(TestCase):
+    fixtures = ["test.json"]
+
+    def test_serve_query_count(self):
+        request = get_dummy_request()
+        Site.find_for_request(request)
+        page, args, kwargs = Page.route_for_request(request, request.path)
+        with mock.patch.object(page, "serve", wraps=page.serve) as m:
+            with self.assertNumQueries(0):
+                serve(request, "/")
+            m.assert_called_once_with(request, *args, **kwargs)
+
+    def test_process_view_by_page_query_count(self):
+        expected_query_count = 3
+        site = Site.objects.get()
+        page = site.root_page.add_child(
+            instance=SimplePage(title="Simple page", slug="simple", content="Simple")
+        )
+        with mock.patch.object(
+            Page, "route_for_request", wraps=Page.route_for_request
+        ) as m:
+            with self.modify_settings(
+                MIDDLEWARE={
+                    "prepend": "wagtail.test.middleware.SimplePageViewInterceptorMiddleware"
+                }
+            ):
+                with self.assertNumQueries(expected_query_count):
+                    response_a = self.client.get("/simple/")
+                self.assertEqual(
+                    response_a.content,
+                    b'\n\n\n\n<!DOCTYPE HTML>\n<html lang="en" dir="ltr">\n    <head>\n        <title>Simple page</title>\n    </head>\n    <body>\n        \n        <h1>Simple page</h1>\n        \n    <h2>Simple page</h2>\n\n    </body>\n</html>\n',
+                )
+                self.assertEqual(m.call_count, 2)
+                page.content = "Intercept me"
+                page.save_revision().publish()
+                m.reset_mock()
+                with self.assertNumQueries(expected_query_count):
+                    # verify the same number of queries are used when the
+                    # middleware activates to demonstrate Page.route_for_request()
+                    # prevents extra database queries for serving pages
+                    response_b = self.client.get("/simple/")
+                self.assertEqual(response_b.content, b"Intercepted")
+                self.assertEqual(m.call_count, 1)

wagtail/views.py

@@ -6,19 +6,15 @@ from django.utils.http import url_has_allowed_host_and_scheme
 
 from wagtail import hooks
 from wagtail.forms import PasswordViewRestrictionForm
-from wagtail.models import Page, PageViewRestriction, Site
+from wagtail.models import Page, PageViewRestriction
 
 
 def serve(request, path):
-    # we need a valid Site object corresponding to this request in order to proceed
+    route_result = Page.route_for_request(request, path)
-    site = Site.find_for_request(request)
+    if route_result is None:
-    if not site:
         raise Http404
-
+    else:
-    path_components = [component for component in path.split("/") if component]
+        page, args, kwargs = route_result
-    page, args, kwargs = site.root_page.localized.specific.route(
-        request, path_components
-    )
 
     for fn in hooks.get_hooks("before_serve_page"):
         result = fn(page, request, args, kwargs)