Adds bulk-import option for "import-records" task (#12792)

* skip plugin events if importing or migrating data

* Add bulkloaddata option

* Skip signals if importing

* Improve bulkloaddata command

* Optionally rebuild thumbnails

* enhancements for import_records task

* cache natural key references in bulkloaddata

* wrap export_records in @state_logger

* Reduce file size of exported data

* Added docs

* Test bulk workflow as part of CI

* Add progress bar for data import

* fix for import workflow bug

* Separately test bulk import workflow

* Exercise --prettify option

* Additional CI checks for content excludes

* Allow plugin loading for list_apps

* Additional CI unit tests

* Test for importing with conflicting records

* path fixes

* Adjust test conditions

src/backend/InvenTree/InvenTree/management/commands/bulkloaddata.py

@@ -0,0 +1,213 @@
+"""Custom management command to load fixtures faster using bulk_create()."""
+
+import time
+from contextlib import contextmanager
+
+from django.core.management.base import CommandError
+from django.core.management.commands.loaddata import Command as LoadDataCommand
+from django.core.serializers import base as serializers_base
+from django.db import DatabaseError, IntegrityError, connections, router
+
+import structlog
+from tqdm import tqdm
+
+logger = structlog.get_logger('inventree')
+
+DEFAULT_BATCH_SIZE = 500
+
+# Number of records to process between progress bar updates. Calling tqdm's
+# update() for every single record adds measurable overhead of its own (well
+# beyond just the display refresh) once there are a million or more of them -
+# so update it in batches instead.
+PROGRESS_UPDATE_INTERVAL = 100
+
+
+class Command(LoadDataCommand):
+    """Load fixtures using bulk_create() for improved performance.
+
+    Behaves like the built-in 'loaddata' command, with two differences to be
+    aware of:
+
+    - pre_save / post_save signals are not sent, and Model.save() / full_clean()
+      are bypassed entirely (this is a Django bulk_create() limitation).
+    - Multi-table inheritance is not supported by bulk_create() and will fail
+      loudly rather than being silently mishandled. Natural-key foreign key /
+      many-to-many resolution *is* supported (falling back to an individual,
+      non-bulk save for any row that needs it - see save_obj()) and is further
+      sped up by caching each resolved natural key for the life of the command
+      (see _cached_natural_keys()), since 'export_records' uses
+      --natural-foreign and a large fixture can have many rows referencing the
+      same handful of natural-keyed objects (e.g. stock.StockItemTracking.user
+      -> auth.User) - without caching, each one costs a separate DB query.
+
+    Based on the django forum thread:
+    - https://forum.djangoproject.com/t/feature-proposal-faster-fixture-loading-via-loaddata-command/36972/21
+    """
+
+    def add_arguments(self, parser):
+        """Add bulkloaddata-specific arguments, on top of loaddata's own."""
+        super().add_arguments(parser)
+        parser.add_argument(
+            '--batch-size',
+            type=int,
+            default=DEFAULT_BATCH_SIZE,
+            help=f'Number of records per bulk_create() batch (default: {DEFAULT_BATCH_SIZE})',
+        )
+        parser.add_argument(
+            '--ignore-conflicts',
+            action='store_true',
+            help='Skip records that violate a unique constraint, instead of raising an error',
+        )
+
+    def handle(self, *fixture_labels, **options):
+        """Store bulk-loading options before delegating to the base command."""
+        self.batch_size = options['batch_size']
+        self.ignore_conflicts = options['ignore_conflicts']
+        self.pending_objs = {}
+        self.pending_progress = 0
+
+        connection = connections[options['database']]
+        self.query_count = 0
+
+        def count_queries(execute, sql, params, many, context):
+            """Count every query executed against this connection, without the overhead of recording each query's SQL text (unlike e.g. CaptureQueriesContext)."""
+            self.query_count += 1
+            return execute(sql, params, many, context)
+
+        start_time = time.monotonic()
+
+        with (
+            connection.execute_wrapper(count_queries),
+            self._cached_natural_keys(),
+            tqdm(desc='Importing', unit=' records') as self.progress,
+        ):
+            super().handle(*fixture_labels, **options)
+            self.progress.update(self.pending_progress)
+
+        elapsed = time.monotonic() - start_time
+
+        if self.verbosity >= 1:
+            self.stdout.write(
+                f'Executed {self.query_count} database queries in {elapsed:.2f}s'
+            )
+
+    @contextmanager
+    def _cached_natural_keys(self):
+        """Cache natural-key foreign key resolutions for the duration of this block.
+
+        Django's deserializer (deserialize_fk_value) issues a fresh DB query
+        every time it resolves a natural-key FK reference, with no caching of
+        its own. A fixture with many rows referencing the same handful of
+        natural-keyed objects (e.g. thousands of stock.StockItemTracking rows
+        all pointing at a few auth.User accounts) would otherwise cost one
+        query per row instead of one query per distinct value.
+        """
+        cache = {}
+        original = serializers_base.deserialize_fk_value
+
+        def cached_deserialize_fk_value(
+            field, field_value, using, handle_forward_references
+        ):
+            default_manager = field.remote_field.model._default_manager
+
+            is_natural_key = (
+                field_value is not None
+                and hasattr(default_manager, 'get_by_natural_key')
+                and hasattr(field_value, '__iter__')
+                and not isinstance(field_value, str)
+            )
+
+            if not is_natural_key:
+                # Plain (non natural-key) FK values never reach a DB query in
+                # the first place - nothing to cache, delegate as normal
+                return original(field, field_value, using, handle_forward_references)
+
+            cache_key = (field.remote_field.model, using, tuple(field_value))
+
+            if cache_key in cache:
+                return cache[cache_key]
+
+            value = original(field, field_value, using, handle_forward_references)
+
+            # Only cache a fully-resolved value - a deferred lookup (the
+            # referenced object doesn't exist yet) may well succeed on a later
+            # call, once that object has actually been saved.
+            if value is not serializers_base.DEFER_FIELD:
+                cache[cache_key] = value
+
+            return value
+
+        serializers_base.deserialize_fk_value = cached_deserialize_fk_value
+        try:
+            yield
+        finally:
+            serializers_base.deserialize_fk_value = original
+
+    def save_obj(self, obj):
+        """Buffer an object for bulk insertion, instead of saving it immediately."""
+        self.pending_progress += 1
+        if self.pending_progress >= PROGRESS_UPDATE_INTERVAL:
+            self.progress.update(self.pending_progress)
+            self.pending_progress = 0
+
+        if (
+            obj.object._meta.app_config in self.excluded_apps
+            or type(obj.object) in self.excluded_models
+        ):
+            return False
+
+        if not router.allow_migrate_model(self.using, obj.object.__class__):
+            return False
+
+        self.models.add(obj.object.__class__)
+
+        if obj.deferred_fields:
+            # This object has an unresolved forward reference (e.g. a natural-key
+            # FK to an object that has not been saved yet - 'export_records' uses
+            # --natural-foreign, and e.g. auth.User defines a natural_key(), so
+            # this does occur in practice). It cannot be bulk_create()'d as-is, since
+            # the deferred field would be written as blank/null. The base loaddata()
+            # command already resolves and saves objects like this individually,
+            # via save_deferred_fields(), once every fixture file has been buffered
+            # (see 'handle' -> 'loaddata') - so just hand it off for that, rather
+            # than also bulk-inserting it here with the field left unresolved.
+            self.objs_with_deferred_fields.append(obj)
+        else:
+            self.pending_objs.setdefault(obj.object.__class__, []).append(obj)
+
+        return True
+
+    def flush_pending(self):
+        """Bulk-create every object buffered so far, grouped by model."""
+        for model, objs in self.pending_objs.items():
+            try:
+                model._default_manager.db_manager(self.using).bulk_create(
+                    [obj.object for obj in objs],
+                    batch_size=self.batch_size,
+                    ignore_conflicts=self.ignore_conflicts,
+                )
+            except (DatabaseError, IntegrityError, ValueError) as e:
+                e.args = (
+                    f'Could not bulk-create {len(objs)} object(s) of {model._meta.label}: {e}',
+                )
+                raise
+
+            # bulk_create() cannot populate many-to-many relations - apply them here,
+            # same as DeserializedObject.save() does for the non-bulk path.
+            for obj in objs:
+                if obj.m2m_data:
+                    for accessor_name, values in obj.m2m_data.items():
+                        getattr(obj.object, accessor_name).set(values)
+                    obj.m2m_data = None
+
+        self.pending_objs = {}
+
+    def load_label(self, fixture_label):
+        """Load one fixture label, then flush the records it buffered."""
+        super().load_label(fixture_label)
+        try:
+            self.flush_pending()
+        except Exception as e:
+            if not isinstance(e, CommandError):
+                e.args = (f"Problem installing fixture '{fixture_label}': {e}",)
+            raise

src/backend/InvenTree/InvenTree/ready.py

@@ -49,7 +49,7 @@ def isWaitingForDatabase():
 
 def isImportingData():
     """Returns True if the database is currently importing (or exporting) data, e.g. 'loaddata' command is performed."""
-    return any(x in sys.argv for x in ['flush', 'loaddata', 'dumpdata'])
+    return any(x in sys.argv for x in ['flush', 'loaddata', 'bulkloaddata', 'dumpdata'])
 
 
 def isRunningMigrations():
@@ -285,7 +285,7 @@ def canAppAccessDatabase(
         excluded_commands.append('test')
 
     if not allow_plugins:
-        excluded_commands.extend(['collectplugins'])
+        excluded_commands.extend(['collectplugins', 'list_apps'])
 
     return all(cmd not in sys.argv for cmd in excluded_commands)
 

src/backend/InvenTree/InvenTree/test_commands.py

@@ -7,6 +7,7 @@ from pathlib import Path
 from django.conf import settings
 from django.contrib.auth.models import User
 from django.core.management import call_command
+from django.db import IntegrityError
 from django.test import TestCase
 
 from opentelemetry.instrumentation.sqlite3 import SQLite3Instrumentor
@@ -116,6 +117,50 @@ class CommandTestCase(TestCase):
         self.assertEqual(output, 'done')
         self.assertEqual(my_admin3.authenticator_set.all().count(), 0)
 
+    def test_bulkloaddata(self):
+        """Test the bulkloaddata command."""
+        from django.contrib.contenttypes.models import ContentType
+        from django.core import serializers
+
+        # ContentType has no custom signals and a real unique_together constraint
+        # (app_label, model), making it a convenient, side-effect-free test model.
+        entries = [
+            ContentType.objects.create(app_label='bulkloaddata_test', model=f'model{i}')
+            for i in range(5)
+        ]
+        pks = [e.pk for e in entries]
+        data = serializers.serialize('json', entries)
+        ContentType.objects.filter(pk__in=pks).delete()
+
+        tmp_file = get_testfolder_dir().joinpath('bulkloaddata_test.json')
+        tmp_file.write_text(data, encoding='utf-8')
+
+        try:
+            # Basic load - all records should be recreated
+            call_command('bulkloaddata', str(tmp_file), verbosity=0)
+            self.assertEqual(ContentType.objects.filter(pk__in=pks).count(), 5)
+
+            # Re-loading without --ignore-conflicts should raise (unique app_label/model)
+            with self.assertRaises(IntegrityError):
+                call_command('bulkloaddata', str(tmp_file), verbosity=0)
+
+            # Re-loading with --ignore-conflicts should succeed, without duplicating rows
+            call_command(
+                'bulkloaddata', str(tmp_file), verbosity=0, ignore_conflicts=True
+            )
+            self.assertEqual(ContentType.objects.filter(pk__in=pks).count(), 5)
+
+            # A small batch size should still load every record correctly
+            ContentType.objects.filter(pk__in=pks).delete()
+            call_command('bulkloaddata', str(tmp_file), verbosity=0, batch_size=2)
+            models = set(
+                ContentType.objects.filter(pk__in=pks).values_list('model', flat=True)
+            )
+            self.assertEqual(models, {e.model for e in entries})
+        finally:
+            ContentType.objects.filter(pk__in=pks).delete()
+            tmp_file.unlink(missing_ok=True)
+
     def test_backup_metadata(self):
         """Test the backup metadata functions."""
         from InvenTree.backup import (

src/backend/InvenTree/order/models.py

@@ -4287,6 +4287,9 @@ class TransferOrderAllocation(models.Model):
 
 def _touch_order_updated_at(instance):
     """Bump updated_at on the parent order without triggering a full save."""
+    if InvenTree.ready.isRunningMigrations() or InvenTree.ready.isImportingData():
+        # Do not touch the order during migrations or data import
+        return
     if not InvenTree.ready.canAppAccessDatabase(allow_test=True):
         return
     instance.order.__class__.objects.filter(pk=instance.order_id).update(

src/backend/InvenTree/plugin/base/event/events.py

@@ -14,7 +14,7 @@ from opentelemetry import trace
 
 import InvenTree.exceptions
 from common.settings import get_global_setting
-from InvenTree.ready import canAppAccessDatabase, isImportingData
+from InvenTree.ready import canAppAccessDatabase, isImportingData, isRunningMigrations
 from InvenTree.tasks import bulk_offload_task, offload_task
 from plugin import PluginMixinEnum
 from plugin.registry import registry
@@ -227,6 +227,10 @@ def process_event(plugin_slug, event, *args, **kwargs):
     This function is run by the background worker process.
     This function may queue multiple functions to be handled by the background worker.
     """
+    if isRunningMigrations() or isImportingData():
+        # Do not trigger events during migrations or data import
+        return
+
     plugin = registry.get_plugin(plugin_slug, active=True)
 
     if plugin is None:  # pragma: no cover
@@ -293,6 +297,10 @@ def allow_table_event(table_name):
 @receiver(post_save)
 def after_save(sender, instance, created, **kwargs):
     """Trigger an event whenever a database entry is saved."""
+    if isRunningMigrations() or isImportingData():
+        # Do not trigger events during migrations or data import
+        return
+
     table = sender.objects.model._meta.db_table
 
     instance_id = getattr(instance, 'id', None)
@@ -312,6 +320,10 @@ def after_save(sender, instance, created, **kwargs):
 @receiver(post_delete)
 def after_delete(sender, instance, **kwargs):
     """Trigger an event whenever a database entry is deleted."""
+    if isRunningMigrations() or isImportingData():
+        # Do not trigger events during migrations or data import
+        return
+
     table = sender.objects.model._meta.db_table
 
     if not allow_table_event(table):