Adds bulk-import option for "import-records" task

.github/workflows/import_export.yaml

@@ -112,7 +112,7 @@ jobs:
           test -f /home/runner/work/InvenTree/test_inventree_db.sqlite3 || (echo "Sqlite database not created" && exit 1)
       - name: Import Sqlite Dataset
         run: |
-          invoke import-records -c -f ${{ env.DATA_FILE }} --strict
+          invoke import-records -c -f ${{ env.DATA_FILE }} --strict --bulk
           cd src/backend/InvenTree && python manage.py check_dummy_data
       - name: Export Sqlite Dataset
         run: |

docs/docs/start/migrate.md

@@ -31,6 +31,9 @@ This will create JSON file at the specified location which contains all database
 !!! info "Specifying filename"
     The filename of the exported file can be specified using the `-f` option. To see all available options, run `invoke export-records --help`
 
+!!! info "File Size"
+    By default the exported file is written as compact JSON, to keep its size down. Add the `-p` / `--prettify` option to pretty-print the output with indentation, which is easier to read manually but can roughly double the file size for a large database.
+
 ```
 {{ invoke_commands('export-records --help') }}
 ```
@@ -67,6 +70,16 @@ invoke import-records -c -f data.json
 !!! warning "Character Encoding"
 	If the character encoding of the data file does not exactly match the target database, the import operation may not succeed. In this case, some manual editing of the database JSON file may be required.
 
+!!! tip "Faster Imports"
+    For very large datasets, add the `-b` / `--bulk` option to use a faster import path (the `bulkloaddata` management command) which inserts records in large batches and skips per-record signal processing, rather than saving each record individually:
+
+    ```
+    invoke import-records -c -b -f data.json
+    ```
+
+!!! tip "Strict Metadata Validation"
+    By default, a mismatch between the source and target InvenTree versions (see the "Database Versions" warning above) only produces a warning, and the import continues. Add the `-s` / `--strict` option to fail immediately instead, if you want to guarantee the versions match exactly before any data is written.
+
 ```
 {{ invoke_commands('import-records --help') }}
 ```
@@ -220,6 +233,9 @@ When running the `import-records` command, the import process will also attempt
 2. The plugin *version* must be the same in both installations. If the plugin version is different, then the database schema may be different, and thus the import process may fail.
 3. The InvenTree software version must be the same in both installations. If the InvenTree version is different, then the database schema may be different, and thus the import process may fail.
 
+!!! tip "Skipping Missing Data"
+    If the import file references a plugin (or any other model) that cannot be matched to the current installation - for example, condition 1 above is not met - add the `-i` / `--ignore-nonexistent` option to skip those records instead of failing the entire import.
+
 If all of the above conditions are met, then the plugin data *should* be imported correctly into the new database. To achieve this reliably, the following process steps are implemented in the `import-records` command:
 
 1. The database is cleaned of all existing records (if the `-c` option is used).

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():

src/backend/InvenTree/InvenTree/test_commands.py

@@ -7,6 +7,7 @@
 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 @@ def get_dummyuser(uname='admin'):
         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 @@ def complete_allocation(self, user):
 
 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 @@
 
 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):