Be more defensive with fetching schema version from Django migration records

wagtail_localize/models.py

@@ -8,7 +8,7 @@
 from django.contrib.contenttypes.models import ContentType
 from django.core.exceptions import ValidationError
 from django.core.serializers.json import DjangoJSONEncoder
-from django.db import models, transaction
+from django.db import OperationalError, models, transaction
 from django.db.migrations.recorder import MigrationRecorder
 from django.db.models import (
     Case,
@@ -126,17 +126,21 @@ def get_edit_url(instance):
         )
 
 
-def get_schema_version(app_label):
+def get_schema_version(app_label: str) -> str:
     """
     Returns the name of the last applied migration for the given app label.
     """
-    migration = (
-        MigrationRecorder.Migration.objects.filter(app=app_label)
-        .order_by("applied")
-        .last()
-    )
-    if migration:
-        return migration.name
+    try:
+        if migration := (
+            MigrationRecorder.Migration.objects.filter(app=app_label)
+            .order_by("applied")
+            .last()
+        ):
+            return migration.name
+    except OperationalError:
+        ...
+
+    return ""
 
 
 class TranslatableObjectManager(models.Manager):
@@ -378,7 +382,7 @@ def get_or_create_from_instance(cls, instance):
                 "locale": instance.locale,
                 "object_repr": str(instance)[:200],
                 "content_json": content_json,
-                "schema_version": get_schema_version(instance._meta.app_label) or "",
+                "schema_version": get_schema_version(instance._meta.app_label),
                 "last_updated_at": timezone.now(),
             },
         )
@@ -436,7 +440,7 @@ def update_or_create_from_instance(cls, instance):
                 "locale": instance.locale,
                 "object_repr": str(instance)[:200],
                 "content_json": content_json,
-                "schema_version": get_schema_version(instance._meta.app_label) or "",
+                "schema_version": get_schema_version(instance._meta.app_label),
                 "last_updated_at": timezone.now(),
             },
         )
@@ -460,7 +464,7 @@ def update_from_db(self):
             serializable_data = get_serializable_data_for_fields(instance)
             self.content_json = json.dumps(serializable_data, cls=DjangoJSONEncoder)
 
-        self.schema_version = get_schema_version(instance._meta.app_label) or ""
+        self.schema_version = get_schema_version(instance._meta.app_label)
         self.object_repr = str(instance)[:200]
         self.last_updated_at = timezone.now()
 

wagtail_localize/tests/test_translation_model.py

@@ -2,6 +2,9 @@
 
 import polib
 
+from django.conf import settings
+from django.db import OperationalError
+from django.db.migrations.recorder import MigrationRecorder
 from django.test import TestCase, override_settings
 from django.utils import timezone
 from wagtail.models import Locale, Page
@@ -23,6 +26,7 @@
     TranslationSource,
     UnknownContext,
     UnknownString,
+    get_schema_version,
 )
 from wagtail_localize.segments import RelatedObjectSegmentValue
 from wagtail_localize.strings import StringValue
@@ -1108,3 +1112,30 @@ def test_sync_view_restrictions_raises_exception_for_snippets(self):
 
         with self.assertRaises(NoViewRestrictionsError):
             source.sync_view_restrictions(snippet, snippet)
+
+
+class TestHelpers(TestCase):
+    def test_get_schema_version_with_no_migrations(self):
+        self.assertEqual(get_schema_version("test.TestSnippet"), "")
+
+    def test_get_schema_version_with_a_migration(self):
+        self.assertEqual(get_schema_version("test.TestSnippet"), "")
+        MigrationRecorder.Migration.objects.create(
+            app="test.TestSnippet",
+            name="0001_initial",
+        )
+        self.assertEqual(get_schema_version("test.TestSnippet"), "0001_initial")
+
+    @mock.patch("wagtail_localize.models.MigrationRecorder.Migration.objects.filter")
+    def test_get_schema_version_returns_empty_string_on_operational_error(
+        self, mocked_filter
+    ):
+        mocked_filter.side_effect = OperationalError
+
+        self.assertEqual(get_schema_version("test.TestSnippet"), "")
+
+    def test_get_schema_version_with_TEST_MIGRATE_false(self):
+        databases = settings.DATABASES.copy()
+        databases["TEST"] = {"MIGRATE": False}
+        with override_settings(DATABASES=databases):
+            self.assertEqual(get_schema_version("test.TestSnippet"), "")