Support long-running migrations starting with numbers

kobo/apps/long_running_migrations/models.py

@@ -1,5 +1,5 @@
-import importlib
 import os
+from importlib.util import module_from_spec, spec_from_file_location
 
 from django.conf import settings
 from django.core.exceptions import SuspiciousOperation
@@ -47,14 +47,7 @@ def execute(self):
         if self.status == LongRunningMigrationStatus.COMPLETED:
             return
 
-        base_import = self.LONG_RUNNING_MIGRATIONS_DIR.replace('/', '.')
-        try:
-            module = importlib.import_module('.'.join([base_import, self.name]))
-        except ModuleNotFoundError as e:
-            logging.error(
-                f'LongRunningMigration.execute(), '
-                f'failed to import task module: {str(e)}'
-            )
+        if not (module := self._load_module()):
             return
 
         self.status = LongRunningMigrationStatus.IN_PROGRESS
@@ -84,3 +77,32 @@ def save(self, **kwargs):
             if not os.path.exists(file_path):
                 raise ValueError('Task does not exist in tasks directory')
         super().save(**kwargs)
+
+    def _load_module(self):
+        """
+        This function allows you to load a Python module from a file path even if
+        the module's name does not follow Python's standard naming conventions
+        (e.g., starting with numbers or containing special characters). Normally,
+        Python identifiers must adhere to specific rules, but this method bypasses
+        those restrictions by dynamically creating a module from its file.
+        """
+        module_path = f'{self.LONG_RUNNING_MIGRATIONS_DIR}/{self.name}.py'
+        if not os.path.exists(f'{settings.BASE_DIR}/{module_path}'):
+            logging.error(
+                f'LongRunningMigration._load_module():'
+                f'File not found `{module_path}`'
+            )
+            return
+
+        spec = spec_from_file_location(self.name, module_path)
+        try:
+            module = module_from_spec(spec)
+        except (ModuleNotFoundError, AttributeError):
+            logging.error(
+                f'LongRunningMigration._load_module():'
+                f'Failed to import migration module `{self.name}`'
+            )
+            return
+
+        spec.loader.exec_module(module)
+        return module

kobo/apps/long_running_migrations/tests/test_case.py

@@ -43,6 +43,7 @@ def test_sample_failure(self):
         self.assertEqual(migration.status, LongRunningMigrationStatus.FAILED)
 
     def test_not_updated_worker(self):
+        # simulate not updated worker with a wrong name
         migrations = LongRunningMigration.objects.bulk_create(
             [LongRunningMigration(name='foo')]
         )
@@ -68,6 +69,10 @@ def setUp(self):
         self.patcher.start()
         self.migration = LongRunningMigration.objects.create(name='sample_task')
 
+        # Remove real existing long-running migrations
+        LongRunningMigration.objects.exclude(pk=self.migration.pk).delete()
+        assert LongRunningMigration.objects.count() == 1
+
     def tearDown(self):
         self.patcher.stop()