Improve index update

src/core/management/commands/update_index_async.py

@@ -1,4 +1,6 @@
+from django.core.cache import cache
 from django.core.management.base import BaseCommand
+from wagtail.search.management.commands.update_index import Command
 
 
 class Command(BaseCommand):
@@ -7,8 +9,41 @@ class Command(BaseCommand):
         " command"
     )
 
+    def add_arguments(self, parser):
+        parser.add_argument(
+            "--force",
+            help="Force the update of the search index",
+            dest="force",
+            default=False,
+        )
+
     def handle(self, *args, **options):
         from core.tasks import update_search_index
+        from extended_search.management.commands.create_index_mapping_json import (
+            get_indexed_mapping_dict,
+        )
+
+        force = options["force"]
+        perform_update = force
+
+        if not force:
+            # Get the current search mapping hash
+            sm_hash = cache.get("search_mapping_hash")
+
+            # Get the search mapping dict
+            sm_dict = get_indexed_mapping_dict()
+            # Get the hash of the current search mapping dict
+            new_sm_hash = hash(frozenset(sm_dict.items()))
+
+            # If the hash of the current search mapping dict is different from the
+            # hash of the search mapping dict that was stored in the cache, then
+            # update the search index.
+            if sm_hash != new_sm_hash:
+                cache.set("search_mapping_hash", new_sm_hash)
+                perform_update = True
+
+        if not perform_update:
+            return
 
         update_search_index.delay()
 

src/core/tasks.py

@@ -28,10 +28,6 @@ def ingest_uk_staff_locations(self):
 
 @celery_app.task(bind=True)
 def update_search_index(self):
-    # Run update_index --schema-only
-    call_command("update_index", schema_only=True)
-
-    # Run update_index
     call_command("update_index")
 
 

src/extended_search/management/commands/create_index_mapping_json.py

@@ -0,0 +1,34 @@
+from pathlib import Path
+
+from django.core.management.base import BaseCommand
+from wagtail.search.backends import get_search_backend
+
+from extended_search.index import get_indexed_models
+
+
+# Path: src/extended_search/management/commands/create_index_fields_json.py
+JSON_FILE = Path(__file__).parent / "indexed_mapping.json"
+
+
+def get_indexed_mapping_dict():
+    """
+    Return a dictionary of indexed models and their fields
+    Ignoring some models that we don't care about.
+    """
+
+    search_backend = get_search_backend()
+
+    return {
+        str(model): search_backend.mapping_class(model).get_mapping()
+        for model in get_indexed_models()
+    }
+
+
+class Command(BaseCommand):
+    help = "Create test JSON containing the mapping for all indexed models"
+
+    def handle(self, *args, **options):
+        import json
+
+        with open(JSON_FILE, "w") as f:
+            json.dump(get_indexed_mapping_dict(), f, indent=4)