[refactor] Bulk actions (#12359)

* Add helper to bulk offload task

* bulk offload events

* Add unit test for bulk_offload_task

* Add context wrapper for bulk event creation inside looped funcs

* Add batching for stock tracking events

* Context wrapper for capturing offloaded tasks

* apply batching to other stock adjustment endpoints

* Add benchmarking tests for other API endpoints

* Benchmark test for serialize stock

* Reduce commet verbosity

* Apply benchmark to PO receive

* Refactor benchmarking

* PrefetchedPrimaryKeyRelatedField

- Provides O(1) lookup for serializer validation
- Significant improvements for large datasets

* Adjust unit test code

* Check for correct order of tracking operations

src/backend/InvenTree/InvenTree/fields.py

@@ -14,6 +14,7 @@ from djmoney.models.fields import MoneyField as ModelMoneyField
 from djmoney.models.validators import MinMoneyValidator
 from rest_framework.fields import URLField as RestURLField
 from rest_framework.fields import empty
+from rest_framework.relations import PrimaryKeyRelatedField
 
 import InvenTree.helpers
 import InvenTree.ready
@@ -48,6 +49,42 @@ class InvenTreeRestURLField(RestURLField):
         return super().run_validation(data=data)
 
 
+class PrefetchedPrimaryKeyRelatedField(PrimaryKeyRelatedField):
+    """A PrimaryKeyRelatedField which resolves against a pre-fetched {pk: instance} map.
+
+    PrimaryKeyRelatedField normally issues one .get() query per list entry when used
+    inside a many=True nested serializer - for large lists (hundreds of related objects)
+    that becomes an O(n) query cost just to validate the request. The parent serializer
+    should instead bulk-fetch all referenced objects in a single query and stash the
+    {pk: instance} map in self.context[cache_key] (typically from an overridden
+    to_internal_value()); this field then does an O(1) dict lookup instead of hitting
+    the database.
+
+    Falls back to the default per-item query if no cache has been populated (or the pk
+    is missing from it), so this field remains safe to use standalone - e.g. in tests
+    constructing the child serializer directly, or for a pk that's genuinely invalid.
+    """
+
+    def __init__(self, cache_key: str, **kwargs):
+        """Store the context key under which the parent serializer stashes its prefetch cache."""
+        self.cache_key = cache_key
+        super().__init__(**kwargs)
+
+    def to_internal_value(self, data):
+        """Resolve 'data' (a raw pk value) against the prefetch cache, if available."""
+        cache = self.context.get(self.cache_key)
+
+        try:
+            pk = int(data)
+        except (TypeError, ValueError):
+            pk = None
+
+        if not cache or pk not in cache:
+            return super().to_internal_value(data)
+
+        return cache[pk]
+
+
 class InvenTreeURLField(models.URLField):
     """Custom URL field which has custom scheme validators."""
 

src/backend/InvenTree/InvenTree/tasks.py

@@ -1,10 +1,13 @@
 """Functions for tasks and a few general async tasks."""
 
+import contextvars
 import json
 import os
 import re
 import warnings
+from collections import defaultdict
 from collections.abc import Callable
+from contextlib import contextmanager
 from dataclasses import dataclass
 from datetime import datetime, timedelta
 from typing import Optional
@@ -13,7 +16,7 @@ from django.conf import settings
 from django.contrib.auth import get_user_model
 from django.core.exceptions import AppRegistryNotReady, ValidationError
 from django.core.management import call_command
-from django.db import DEFAULT_DB_ALIAS, connections
+from django.db import DEFAULT_DB_ALIAS, connections, transaction
 from django.db.migrations.executor import MigrationExecutor
 from django.db.utils import NotSupportedError, OperationalError, ProgrammingError
 from django.utils import timezone
@@ -201,6 +204,86 @@ def check_existing_task(taskname, group: str, *args, **kwargs) -> Optional[str]:
     return task_id
 
 
+# Context-local batch of pending offload_task() calls (see batch_offload_tasks())
+_task_batch: contextvars.ContextVar = contextvars.ContextVar('task_batch', default=None)
+
+
+class TaskBatch:
+    """Collects offload_task() calls made within a batch_offload_tasks() scope.
+
+    Entries are grouped by (taskname, group, force_async), so that each distinct
+    combination triggered within the batch is flushed via its own bulk_offload_task() call.
+    """
+
+    def __init__(self):
+        """Initialize an empty batch."""
+        self.entries: dict[tuple, list] = defaultdict(list)
+
+    def add(
+        self, taskname, group: str, force_async: bool, args: tuple, kwargs: dict
+    ) -> None:
+        """Record a single offload_task() call against this batch."""
+        self.entries[taskname, group, force_async].append((args, kwargs))
+
+    def flush(self) -> None:
+        """Fire a bulk_offload_task() call for each (taskname, group, force_async) group collected so far."""
+        entries, self.entries = self.entries, defaultdict(list)
+
+        for (taskname, group, force_async), task_entries in entries.items():
+            bulk_offload_task(
+                taskname, task_entries, group=group, force_async=force_async
+            )
+
+
+@contextmanager
+def batch_offload_tasks():
+    """Batch offload_task() calls made within this scope into bulk_offload_task() calls.
+
+    Any offload_task() call made (directly, or indirectly via a nested function call) while
+    this context is active is queued instead of immediately offloaded - *except* for calls
+    which pass force_sync=True, which always run immediately and synchronously as before.
+    Excluding these is necessary because a forced-sync call is relied upon to have completed,
+    with its side effects visible, by the time offload_task() returns control to the caller -
+    deferring it would silently break that contract.
+
+    The queued calls are flushed - grouped by (taskname, group, force_async), one
+    bulk_offload_task() call per group - when the current database transaction commits (or
+    immediately, if no transaction is active). If the transaction is instead rolled back, the
+    queued calls are discarded, rather than being fired for a write that never happened.
+
+    Note: bulk_offload_task() does not perform duplicate-task checking, unlike offload_task()'s
+    default (check_duplicates=True) behavior - queued calls are never deduplicated, regardless
+    of the check_duplicates value passed to offload_task().
+
+    A batched offload_task() call always returns True immediately, rather than a task ID -
+    the actual task ID is not known until the batch is flushed, possibly well after the
+    call returns. Callers which depend on the returned task ID should not use this context.
+
+    Nesting is not supported: a nested batch_offload_tasks() call reuses the outer batch,
+    and only the outermost call schedules a flush.
+
+    This mirrors plugin.base.event.events.batch_events() and stock.models.batch_tracking_entries()
+    - see batch_events()'s docstring for the reasoning behind the on-commit flush and the
+    context-local (rather than parameter-based) design.
+
+    Yields:
+        The current TaskBatch instance
+    """
+    if _task_batch.get() is not None:
+        # Already inside a batch - extend it, rather than creating a nested one
+        yield _task_batch.get()
+        return
+
+    batch = TaskBatch()
+    token = _task_batch.set(batch)
+
+    try:
+        yield batch
+    finally:
+        _task_batch.reset(token)
+        transaction.on_commit(batch.flush)
+
+
 def offload_task(
     taskname,
     *args,
@@ -224,11 +307,18 @@ def offload_task(
     Returns:
         str | bool: Task ID if the task was offloaded, True if ran synchronously, False otherwise
     """
-    from InvenTree.exceptions import log_error
-
     # Extract group information from kwargs
     group = kwargs.pop('group', 'inventree')
 
+    if not force_sync and (batch := _task_batch.get()) is not None:
+        # A batch_offload_tasks() context is active - queue this task rather than
+        # offloading it immediately (force_sync=True calls never reach this branch -
+        # see batch_offload_tasks() for why they are excluded from batching)
+        batch.add(taskname, group, force_async, args, kwargs)
+        return True
+
+    from InvenTree.exceptions import log_error
+
     try:
         import importlib
 
@@ -323,6 +413,96 @@ def offload_task(
     return True
 
 
+def bulk_offload_task(
+    taskname,
+    entries: list,
+    group: str = 'inventree',
+    force_sync: bool = False,
+    force_async: bool = False,
+) -> bool:
+    """Queue the same background task many times, in a single bulk database write.
+
+    Equivalent to calling offload_task() once per (args, kwargs) pair in 'entries', but
+    writes all of the queued tasks to the django-q2 ORM broker table (OrmQ) in a single
+    bulk_create() call, rather than one INSERT per task.
+
+    Note: InvenTree always configures django-q2 to use the ORM broker (see
+    InvenTree.setting.worker.get_worker_config), so this does not need to handle any
+    other broker backend.
+
+    Arguments:
+        taskname: The name of the task to be run, in the format 'app.module.function'
+        entries: List of (args, kwargs) tuples, one per task instance to queue
+        group: The task group to assign to each queued task
+        force_sync: If True, run all tasks synchronously (even if workers are running)
+        force_async: If True, force all tasks to be queued (even if workers are not running)
+
+    Returns:
+        bool: True if the tasks were queued (or run synchronously), False otherwise
+    """
+    if not entries:
+        return False
+
+    try:
+        from django_q.brokers import get_broker
+        from django_q.humanhash import uuid
+        from django_q.models import OrmQ
+        from django_q.signing import SignedPackage
+
+        from InvenTree.status import is_worker_running
+    except AppRegistryNotReady:  # pragma: no cover
+        logger.warning(
+            "Could not offload bulk task '%s' - app registry not ready", taskname
+        )
+        force_sync = True
+    except (OperationalError, ProgrammingError):  # pragma: no cover
+        raise_warning(f"Could not offload bulk task '{taskname}' - database not ready")
+        force_sync = True
+
+    if not force_async and (force_sync or not is_worker_running()):
+        # Workers are not available - fall back to running each task synchronously
+        for args, kwargs in entries:
+            offload_task(
+                taskname,
+                *args,
+                group=group,
+                force_sync=True,
+                check_duplicates=False,
+                **kwargs,
+            )
+
+        return True
+
+    broker = get_broker()
+
+    tasks = []
+
+    for args, kwargs in entries:
+        name, task_id = uuid()
+
+        task = {
+            'id': task_id,
+            'name': name,
+            'func': taskname,
+            'args': args,
+            'kwargs': kwargs,
+            'group': group,
+            'started': timezone.now(),
+        }
+
+        tasks.append(
+            OrmQ(
+                key=broker.list_key or 'inventree',
+                payload=SignedPackage.dumps(task),
+                lock=timezone.now(),
+            )
+        )
+
+    OrmQ.objects.bulk_create(tasks)
+
+    return True
+
+
 def get_queued_task(task_id: str):
     """Find the task in the queue, if it exists.
 

src/backend/InvenTree/InvenTree/test_tasks.py

@@ -6,6 +6,7 @@ from datetime import timedelta
 from django.conf import settings
 from django.contrib.auth.models import User
 from django.core.management import call_command
+from django.db import transaction
 from django.db.utils import NotSupportedError
 from django.test import TestCase
 from django.utils import timezone
@@ -411,3 +412,175 @@ class InvenTreeTaskTests(PluginRegistryMixin, TestCase):
 
         # 20 more tasks should have been added
         self.assertEqual(OrmQ.objects.count(), 41)
+
+    def test_bulk_offload(self):
+        """Test the bulk_offload_task function."""
+        # Start with a blank slate
+        OrmQ.objects.all().delete()
+
+        entries = [
+            ((idx, idx + 1), {'animal': f'animal_{idx}', 'count': idx})
+            for idx in range(10)
+        ]
+
+        # Queuing all 10 tasks should only take a single database write (bulk_create)
+        with self.assertNumQueries(1):
+            result = InvenTree.tasks.bulk_offload_task(
+                'dummy_module.dummy_function', entries, force_async=True
+            )
+
+        self.assertTrue(result)
+        self.assertEqual(OrmQ.objects.count(), 10)
+
+        # Read out the pending tasks, and check that the args / kwargs match
+        queued_tasks = OrmQ.objects.all().order_by('id')
+
+        for task, (args, kwargs) in zip(queued_tasks, entries, strict=True):
+            self.assertEqual(task.func(), 'dummy_module.dummy_function')
+            self.assertEqual(task.group(), 'inventree')
+            self.assertEqual(task.args(), args)
+            self.assertEqual(task.kwargs(), kwargs)
+
+
+class TaskBatchTests(TestCase):
+    """Unit tests for the batch_offload_tasks() context manager."""
+
+    def setUp(self):
+        """Start each test with an empty task queue."""
+        super().setUp()
+        OrmQ.objects.all().delete()
+
+    def test_tasks_queued_and_flushed_on_commit(self):
+        """Tasks offloaded inside batch_offload_tasks() are queued and flushed as one bulk write on commit."""
+        with self.captureOnCommitCallbacks(execute=True):
+            with transaction.atomic(), InvenTree.tasks.batch_offload_tasks():
+                for idx in range(10):
+                    InvenTree.tasks.offload_task(
+                        'dummy_module.dummy_function',
+                        idx,
+                        force_async=True,
+                        animal=f'animal_{idx}',
+                    )
+
+                # Nothing should be queued yet - the batch only flushes on commit
+                self.assertEqual(OrmQ.objects.count(), 0)
+
+        self.assertEqual(OrmQ.objects.count(), 10)
+
+        queued_tasks = OrmQ.objects.all().order_by('id')
+
+        for idx, task in enumerate(queued_tasks):
+            self.assertEqual(task.func(), 'dummy_module.dummy_function')
+            self.assertEqual(task.group(), 'inventree')
+            self.assertEqual(task.args(), (idx,))
+            self.assertEqual(task.kwargs(), {'animal': f'animal_{idx}'})
+
+    def test_tasks_grouped_by_name_and_group(self):
+        """Tasks with different (taskname, group) combinations are flushed as separate bulk writes."""
+        with self.captureOnCommitCallbacks(execute=True):
+            with transaction.atomic(), InvenTree.tasks.batch_offload_tasks():
+                for idx in range(5):
+                    InvenTree.tasks.offload_task(
+                        'dummy_module.task_a', idx, force_async=True, group='alpha'
+                    )
+                for idx in range(3):
+                    InvenTree.tasks.offload_task(
+                        'dummy_module.task_b', idx, force_async=True, group='beta'
+                    )
+
+        self.assertEqual(OrmQ.objects.count(), 8)
+        self.assertEqual(
+            sum(
+                1
+                for t in OrmQ.objects.all()
+                if t.func() == 'dummy_module.task_a' and t.group() == 'alpha'
+            ),
+            5,
+        )
+        self.assertEqual(
+            sum(
+                1
+                for t in OrmQ.objects.all()
+                if t.func() == 'dummy_module.task_b' and t.group() == 'beta'
+            ),
+            3,
+        )
+
+    def test_tasks_discarded_on_rollback(self):
+        """Tasks queued in a batch are discarded, not fired, if the transaction rolls back."""
+        with self.captureOnCommitCallbacks(execute=True):
+            try:
+                with transaction.atomic(), InvenTree.tasks.batch_offload_tasks():
+                    InvenTree.tasks.offload_task(
+                        'dummy_module.dummy_function', force_async=True
+                    )
+                    raise ValueError('boom')
+            except ValueError:
+                pass
+
+        self.assertEqual(OrmQ.objects.count(), 0)
+
+    def test_tasks_outside_batch_fire_immediately(self):
+        """Tasks offloaded outside any batch_offload_tasks() context are unaffected - fired immediately."""
+        InvenTree.tasks.offload_task('dummy_module.dummy_function', force_async=True)
+
+        # No transaction commit or captureOnCommitCallbacks needed - it was never queued
+        self.assertEqual(OrmQ.objects.count(), 1)
+
+    def test_nested_batch_share_one_flush(self):
+        """A nested batch_offload_tasks() call reuses the outer batch, rather than flushing twice."""
+        with self.captureOnCommitCallbacks(execute=True):
+            with transaction.atomic(), InvenTree.tasks.batch_offload_tasks():
+                InvenTree.tasks.offload_task(
+                    'dummy_module.dummy_function', 1, force_async=True
+                )
+                with InvenTree.tasks.batch_offload_tasks():
+                    InvenTree.tasks.offload_task(
+                        'dummy_module.dummy_function', 2, force_async=True
+                    )
+                self.assertEqual(OrmQ.objects.count(), 0)
+
+        self.assertEqual(OrmQ.objects.count(), 2)
+
+    def test_force_sync_excluded_from_batch(self):
+        """force_sync=True calls bypass batch_offload_tasks() entirely and run immediately."""
+        calls = []
+
+        def sync_target():
+            calls.append('ran')
+
+        with self.captureOnCommitCallbacks(execute=True):
+            with transaction.atomic(), InvenTree.tasks.batch_offload_tasks():
+                InvenTree.tasks.offload_task(sync_target, force_sync=True)
+
+                # Ran immediately - not deferred to flush, and never touched the task queue
+                self.assertEqual(calls, ['ran'])
+                self.assertEqual(OrmQ.objects.count(), 0)
+
+                InvenTree.tasks.offload_task(
+                    'dummy_module.dummy_function', force_async=True
+                )
+
+                # The (non-force_sync) async call is queued, not yet in OrmQ
+                self.assertEqual(OrmQ.objects.count(), 0)
+
+        # After commit: only the batched async call produced an OrmQ entry
+        self.assertEqual(OrmQ.objects.count(), 1)
+        self.assertEqual(calls, ['ran'])
+
+    def test_batched_calls_skip_duplicate_check(self):
+        """Unlike individual offload_task() calls, batched calls are never deduplicated."""
+        with self.captureOnCommitCallbacks(execute=True):
+            with transaction.atomic(), InvenTree.tasks.batch_offload_tasks():
+                for _ in range(3):
+                    InvenTree.tasks.offload_task(
+                        'dummy_module.dummy_function_dup',
+                        1,
+                        2,
+                        animal='cat',
+                        force_async=True,
+                    )
+
+        # All 3 identical calls were queued, unlike the non-batched dedup behavior
+        # exercised in InvenTreeTaskTests.test_duplicate_tasks
+        self.assertEqual(OrmQ.objects.count(), 3)

src/backend/InvenTree/InvenTree/unit_test.py

@@ -380,13 +380,23 @@ class TestQueryMixin:
     ):
         """Context manager to check that the number of queries is less than a certain value.
 
+        Arguments:
+            value: The maximum number of queries allowed
+            using: The database connection to use (default = 'default')
+            verbose: If True, print the queries to the console (default = False)
+            url: Optional URL to print in the output (default = None)
+            log_to_file: If True, log the queries to a file (default = False)
+
+        Yields:
+            The CaptureQueriesContext object, which contains the captured queries
+
         Example:
         with self.assertNumQueriesLessThan(10):
             # Do some stuff
         Ref: https://stackoverflow.com/questions/1254170/django-is-there-a-way-to-count-sql-queries-from-an-unit-test/59089020#59089020
         """
         with CaptureQueriesContext(connections[using]) as context:
-            yield  # your test will be run here
+            yield context  # your test will be run here
 
         n = len(context.captured_queries)
 
@@ -492,12 +502,16 @@ class InvenTreeAPITestCase(
 
         expected_code = kwargs.pop('expected_code', None)
         msg = kwargs.pop('msg', None)
-        max_queries = kwargs.pop('max_query_count', self.MAX_QUERY_COUNT)
+        max_query_count = kwargs.pop('max_query_count', self.MAX_QUERY_COUNT)
         max_query_time = kwargs.pop('max_query_time', self.MAX_QUERY_TIME)
+        benchmark = kwargs.pop('benchmark', False)
 
         t1 = time.time()
 
-        with self.assertNumQueriesLessThan(max_queries, url=url):
+        with (
+            self.assertNumQueriesLessThan(max_query_count, url=url) as context,
+            self.captureOnCommitCallbacks(execute=True),
+        ):
             response = method(url, data, **kwargs)
 
         t2 = time.time()
@@ -512,6 +526,11 @@ class InvenTreeAPITestCase(
 
         self.assertLessEqual(dt, max_query_time)
 
+        if benchmark:
+            print(
+                f"Benchmark @ '{url}': {len(context.captured_queries)} queries (of {max_query_count}) in {dt:.4f}s (of {max_query_time}s max)"
+            )
+
         return response
 
     def get(self, url, data=None, expected_code=200, **kwargs):

src/backend/InvenTree/order/serializers.py

@@ -42,6 +42,7 @@ from InvenTree.serializers import (
     NotesFieldMixin,
     OptionalField,
 )
+from InvenTree.tasks import batch_offload_tasks
 from order.status_codes import (
     PurchaseOrderStatusGroups,
     ReturnOrderLineStatus,
@@ -50,6 +51,8 @@ from order.status_codes import (
     TransferOrderStatusGroups,
 )
 from part.serializers import PartBriefSerializer
+from plugin.base.event.events import batch_events
+from stock.models import batch_tracking_entries
 from stock.status_codes import StockStatus
 from users.serializers import OwnerSerializer, UserSerializer
 
@@ -1047,6 +1050,12 @@ class PurchaseOrderReceiveSerializer(serializers.Serializer):
         location = data.get('location', order.destination)
 
         try:
+            with (
+                transaction.atomic(),
+                batch_events(),
+                batch_tracking_entries(),
+                batch_offload_tasks(),
+            ):
                 items = order.receive_line_items(
                     location, items, request.user if request else None
                 )

src/backend/InvenTree/order/test_api.py

@@ -1430,6 +1430,47 @@ class PurchaseOrderReceiveTest(OrderTest):
             line.refresh_from_db()
             self.assertEqual(line.received, line.quantity)
 
+    def test_bulk_receive_query_benchmark(self):
+        """Benchmark: measure the number of DB queries required to receive 100 line items at once."""
+        InvenTreeSetting.set_setting('ENABLE_PLUGINS_EVENTS', True, change_user=None)
+
+        sp = SupplierPart.objects.first()
+
+        po = models.PurchaseOrder.objects.create(
+            reference='PO-BENCHMARK-100', supplier=sp.supplier
+        )
+
+        N_LINES = 100
+
+        models.PurchaseOrderLineItem.objects.bulk_create([
+            models.PurchaseOrderLineItem(order=po, part=sp, quantity=10)
+            for _ in range(N_LINES)
+        ])
+
+        po.place_order()
+
+        url = reverse('api-po-receive', kwargs={'pk': po.pk})
+
+        lines = po.lines.all()
+        location = StockLocation.objects.filter(structural=False).first()
+
+        data = {
+            'items': [
+                {'line_item': line.pk, 'quantity': line.quantity} for line in lines
+            ],
+            'location': location.pk,
+        }
+
+        with self.settings(
+            PLUGIN_TESTING_EVENTS=True, PLUGIN_TESTING_EVENTS_ASYNC=True
+        ):
+            response = self.post(
+                url, data, max_query_count=400, benchmark=True, format='json'
+            )
+
+        self.assertEqual(response.status_code, 201)
+        self.assertEqual(len(response.data), N_LINES)
+
     def test_packaging(self):
         """Test that we can supply a 'packaging' value when receiving items."""
         line_1 = models.PurchaseOrderLineItem.objects.get(pk=1)

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

@@ -1,5 +1,9 @@
 """Functions for triggering and responding to server side events."""
 
+import contextvars
+from collections import defaultdict
+from contextlib import contextmanager
+
 from django.conf import settings
 from django.db import transaction
 from django.db.models.signals import post_delete, post_save
@@ -11,13 +15,79 @@ from opentelemetry import trace
 import InvenTree.exceptions
 from common.settings import get_global_setting
 from InvenTree.ready import canAppAccessDatabase, isImportingData
-from InvenTree.tasks import offload_task
+from InvenTree.tasks import bulk_offload_task, offload_task
 from plugin import PluginMixinEnum
 from plugin.registry import registry
 
 tracer = trace.get_tracer(__name__)
 logger = structlog.get_logger('inventree')
 
+# Active event batch for the current context (see batch_events()), if any
+_event_batch: contextvars.ContextVar = contextvars.ContextVar(
+    'event_batch', default=None
+)
+
+
+class EventBatch:
+    """Collects trigger_event() calls made within a batch_events() scope.
+
+    Entries are grouped by event name, so that each distinct event triggered
+    within the batch is flushed via its own bulk_trigger_event() call.
+    """
+
+    def __init__(self):
+        """Initialize an empty batch."""
+        self.entries: dict[str, list] = defaultdict(list)
+
+    def add(self, event: str, *args, **kwargs) -> None:
+        """Record a single trigger_event() call against this batch."""
+        self.entries[event].append((args, kwargs))
+
+    def flush(self) -> None:
+        """Fire a bulk_trigger_event() call for each event name collected so far."""
+        entries, self.entries = self.entries, defaultdict(list)
+
+        for event, event_entries in entries.items():
+            bulk_trigger_event(event, event_entries)
+
+
+@contextmanager
+def batch_events():
+    """Batch trigger_event() calls made within this scope into bulk_trigger_event() calls.
+
+    Any trigger_event() call made (directly, or indirectly via a nested function call)
+    while this context is active is queued instead of immediately offloaded. The queued
+    events are flushed - grouped by event name, one bulk_trigger_event() call per name -
+    when the current database transaction commits (or immediately, if no transaction is
+    active). If the transaction is instead rolled back, the queued events are discarded,
+    rather than being fired for a write that never happened.
+
+    Nesting is not supported: a nested batch_events() call reuses the outer batch, and
+    only the outermost call schedules a flush.
+
+    This does not change the behavior of code that triggers events *outside* of this
+    context - trigger_event() still fires immediately in that case. This allows existing
+    single-item entrypoints (e.g. StockItem.stocktake()) to be reused unmodified by both
+    single-item and bulk callers: bulk callers simply wrap their loop in batch_events().
+
+    Yields:
+        EventBatch: The current event batch, which can be used to add events directly.
+
+    """
+    if _event_batch.get() is not None:
+        # Already inside a batch - extend it, rather than creating a nested one
+        yield _event_batch.get()
+        return
+
+    batch = EventBatch()
+    token = _event_batch.set(batch)
+
+    try:
+        yield batch
+    finally:
+        _event_batch.reset(token)
+        transaction.on_commit(batch.flush)
+
 
 @tracer.start_as_current_span('trigger_event')
 def trigger_event(event: str, *args, **kwargs) -> None:
@@ -45,6 +115,11 @@ def trigger_event(event: str, *args, **kwargs) -> None:
         logger.debug("Ignoring triggered event '%s' - database not ready", event)
         return
 
+    if (batch := _event_batch.get()) is not None:
+        # A batch_events() context is active - queue this event rather than firing it now
+        batch.add(event, *args, **kwargs)
+        return
+
     logger.debug("Event triggered: '%s'", event)
 
     force_async = kwargs.pop('force_async', True)
@@ -58,6 +133,60 @@ def trigger_event(event: str, *args, **kwargs) -> None:
     offload_task(register_event, event, *args, group='plugin', **kwargs)
 
 
+@tracer.start_as_current_span('bulk_trigger_event')
+def bulk_trigger_event(event: str, entries: list) -> None:
+    """Trigger the same event multiple times, in a single bulk database write.
+
+    Equivalent to calling trigger_event(event, *args, **kwargs) once per (args, kwargs)
+    pair in 'entries', but queues all of the resulting background tasks via a single
+    bulk_offload_task() call, rather than one INSERT per event.
+
+    Arguments:
+        event: The event to trigger
+        entries: List of (args, kwargs) tuples, one per event instance to register
+
+    These events will be stored in the database, and the worker will respond to them later on.
+    """
+    if not entries:
+        return
+
+    if not get_global_setting('ENABLE_PLUGINS_EVENTS', False):
+        # Do nothing if plugin events are not enabled
+        return
+
+    # Ensure event name is stringified
+    event = str(event).strip()
+
+    # Make sure the database can be accessed and is not being tested rn
+    if (
+        not canAppAccessDatabase(allow_shell=True)
+        and not settings.PLUGIN_TESTING_EVENTS
+    ):
+        logger.debug("Ignoring bulk triggered event '%s' - database not ready", event)
+        return
+
+    logger.debug("Bulk event triggered: '%s' (%s entries)", event, len(entries))
+
+    force_async = True
+
+    # If we are running in testing mode, we can enable or disable async processing
+    if settings.PLUGIN_TESTING_EVENTS:
+        force_async = settings.PLUGIN_TESTING_EVENTS_ASYNC
+
+    task_entries = []
+
+    for args, kwargs in entries:
+        kwargs = dict(kwargs)
+        # 'force_async' is a bulk_offload_task() control flag, not event data - it is
+        # resolved once for the whole batch above, so strip any per-entry override
+        kwargs.pop('force_async', None)
+        task_entries.append(((event, *args), kwargs))
+
+    bulk_offload_task(
+        register_event, task_entries, group='plugin', force_async=force_async
+    )
+
+
 @tracer.start_as_current_span('register_event')
 def register_event(event, *args, **kwargs):
     """Register the event with any interested plugins.

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

@@ -0,0 +1,134 @@
+"""Unit tests for event triggering functions."""
+
+from django.db import transaction
+from django.test import TestCase
+
+from django_q.models import OrmQ
+
+from common.models import InvenTreeSetting
+from plugin.base.event.events import batch_events, bulk_trigger_event, trigger_event
+
+
+class BulkEventTriggerTests(TestCase):
+    """Unit tests for the bulk_trigger_event function."""
+
+    def test_bulk_trigger_event(self):
+        """Test that bulk_trigger_event queues events in a single bulk database write."""
+        InvenTreeSetting.set_setting('ENABLE_PLUGINS_EVENTS', True, change_user=None)
+
+        entries = [
+            ((idx, idx + 1), {'animal': f'animal_{idx}', 'count': idx})
+            for idx in range(10)
+        ]
+
+        with self.settings(
+            PLUGIN_TESTING_EVENTS=True, PLUGIN_TESTING_EVENTS_ASYNC=True
+        ):
+            # Start with a blank slate
+            OrmQ.objects.all().delete()
+
+            # Queuing all 10 events should only take two database queries:
+            # one to check the 'ENABLE_PLUGINS_EVENTS' setting, and one bulk_create
+            with self.assertNumQueries(2):
+                bulk_trigger_event('test.event', entries)
+
+        self.assertEqual(OrmQ.objects.count(), 10)
+
+        # Read out the pending tasks, and check that the args / kwargs match
+        queued_tasks = OrmQ.objects.all().order_by('id')
+
+        for task, (args, kwargs) in zip(queued_tasks, entries, strict=True):
+            self.assertEqual(task.func(), 'plugin.base.event.events.register_event')
+            self.assertEqual(task.group(), 'plugin')
+            self.assertEqual(task.args(), ('test.event', *args))
+            self.assertEqual(task.kwargs(), kwargs)
+
+
+class BatchEventsTests(TestCase):
+    """Unit tests for the batch_events() context manager."""
+
+    def setUp(self):
+        """Enable plugin events for all tests in this class."""
+        super().setUp()
+        InvenTreeSetting.set_setting('ENABLE_PLUGINS_EVENTS', True, change_user=None)
+        OrmQ.objects.all().delete()
+
+    def test_events_queued_and_flushed_on_commit(self):
+        """Events triggered inside batch_events() are queued and flushed as one bulk write on commit."""
+        with self.settings(
+            PLUGIN_TESTING_EVENTS=True, PLUGIN_TESTING_EVENTS_ASYNC=True
+        ):
+            with self.captureOnCommitCallbacks(execute=True):
+                with transaction.atomic(), batch_events():
+                    for idx in range(10):
+                        trigger_event('test.event', id=idx, value=f'v{idx}')
+
+                    # Nothing should be queued yet - the batch only flushes on commit
+                    self.assertEqual(OrmQ.objects.count(), 0)
+
+        self.assertEqual(OrmQ.objects.count(), 10)
+
+        queued_tasks = OrmQ.objects.all().order_by('id')
+
+        for idx, task in enumerate(queued_tasks):
+            self.assertEqual(task.args(), ('test.event',))
+            self.assertEqual(task.kwargs(), {'id': idx, 'value': f'v{idx}'})
+
+    def test_events_grouped_by_name(self):
+        """Events with different names in the same batch are flushed as separate bulk writes."""
+        with self.settings(
+            PLUGIN_TESTING_EVENTS=True, PLUGIN_TESTING_EVENTS_ASYNC=True
+        ):
+            with self.captureOnCommitCallbacks(execute=True):
+                with transaction.atomic(), batch_events():
+                    for idx in range(5):
+                        trigger_event('event.a', id=idx)
+                    for idx in range(3):
+                        trigger_event('event.b', id=idx)
+
+        self.assertEqual(OrmQ.objects.count(), 8)
+        self.assertEqual(
+            sum(1 for task in OrmQ.objects.all() if task.args() == ('event.a',)), 5
+        )
+        self.assertEqual(
+            sum(1 for task in OrmQ.objects.all() if task.args() == ('event.b',)), 3
+        )
+
+    def test_events_discarded_on_rollback(self):
+        """Events queued in a batch are discarded, not fired, if the transaction rolls back."""
+        with self.settings(
+            PLUGIN_TESTING_EVENTS=True, PLUGIN_TESTING_EVENTS_ASYNC=True
+        ):
+            with self.captureOnCommitCallbacks(execute=True):
+                try:
+                    with transaction.atomic(), batch_events():
+                        trigger_event('test.event', id=1)
+                        raise ValueError('boom')
+                except ValueError:
+                    pass
+
+        self.assertEqual(OrmQ.objects.count(), 0)
+
+    def test_events_outside_batch_fire_immediately(self):
+        """Events triggered outside any batch_events() context are unaffected - fired immediately."""
+        with self.settings(
+            PLUGIN_TESTING_EVENTS=True, PLUGIN_TESTING_EVENTS_ASYNC=True
+        ):
+            trigger_event('test.event', id=1)
+
+        # No transaction commit or captureOnCommitCallbacks needed - it was never queued
+        self.assertEqual(OrmQ.objects.count(), 1)
+
+    def test_nested_batch_events_share_one_flush(self):
+        """A nested batch_events() call reuses the outer batch, rather than flushing twice."""
+        with self.settings(
+            PLUGIN_TESTING_EVENTS=True, PLUGIN_TESTING_EVENTS_ASYNC=True
+        ):
+            with self.captureOnCommitCallbacks(execute=True):
+                with transaction.atomic(), batch_events():
+                    trigger_event('test.event', id=1)
+                    with batch_events():
+                        trigger_event('test.event', id=2)
+                    self.assertEqual(OrmQ.objects.count(), 0)
+
+        self.assertEqual(OrmQ.objects.count(), 2)

src/backend/InvenTree/plugin/events.py

@@ -1,7 +1,12 @@
 """Import helper for events."""
 
 from generic.events import BaseEventEnum
-from plugin.base.event.events import process_event, register_event, trigger_event
+from plugin.base.event.events import (
+    bulk_trigger_event,
+    process_event,
+    register_event,
+    trigger_event,
+)
 
 
 class PluginEvents(BaseEventEnum):
@@ -11,4 +16,10 @@ class PluginEvents(BaseEventEnum):
     PLUGIN_ACTIVATED = 'plugin_activated'
 
 
-__all__ = ['PluginEvents', 'process_event', 'register_event', 'trigger_event']
+__all__ = [
+    'PluginEvents',
+    'bulk_trigger_event',
+    'process_event',
+    'register_event',
+    'trigger_event',
+]

src/backend/InvenTree/stock/models.py

@@ -2,7 +2,9 @@
 
 from __future__ import annotations
 
+import contextvars
 import os
+from contextlib import contextmanager
 from datetime import timedelta
 from decimal import Decimal, InvalidOperation
 
@@ -2214,6 +2216,10 @@ class StockItem(
         )
 
         if commit:
+            if (batch := _tracking_batch.get()) is not None:
+                # A batch_tracking_entries() context is active - queue rather than save now
+                batch.add(entry)
+            else:
                 entry.save()
 
         return entry
@@ -2954,6 +2960,10 @@ class StockItem(
 
         quantity = max(quantity, 0)
 
+        # No change if the quantity is the same as the current quantity
+        if self.quantity == quantity:
+            return
+
         self.quantity = quantity
 
         if quantity == 0 and self.delete_on_deplete and self.can_delete():
@@ -3324,6 +3334,72 @@ def after_save_stock_item(sender, instance: StockItem, created, **kwargs):
             instance.part.schedule_pricing_update(create=True)
 
 
+# Context-local batch of pending StockItemTracking entries (see batch_tracking_entries())
+_tracking_batch: contextvars.ContextVar = contextvars.ContextVar(
+    'tracking_batch', default=None
+)
+
+
+class TrackingEntryBatch:
+    """Collects StockItemTracking entries created within a batch_tracking_entries() scope."""
+
+    def __init__(self):
+        """Initialize an empty batch."""
+        self.entries: list[StockItemTracking] = []
+
+    def add(self, entry: StockItemTracking) -> None:
+        """Record a single (unsaved) tracking entry against this batch."""
+        self.entries.append(entry)
+
+    def flush(self) -> None:
+        """Bulk-create all entries collected so far, in a single database write."""
+        entries, self.entries = self.entries, []
+
+        if entries:
+            StockItemTracking.objects.bulk_create(entries, batch_size=250)
+
+
+@contextmanager
+def batch_tracking_entries():
+    """Batch StockItemTracking entries created within this scope into a single bulk_create().
+
+    StockItem.add_tracking_entry() saves its entry immediately by default (commit=True).
+    While this context is active, any such call - made directly, or indirectly via a nested
+    function call - is queued instead of saved immediately. The queued entries are flushed via
+    a single bulk_create() when the current transaction commits (or immediately, if no
+    transaction is active). If the transaction is instead rolled back, the queued entries are
+    discarded, rather than being written for a change that never happened.
+
+    Calls that already pass commit=False to add_tracking_entry() are unaffected either way -
+    that contract (the caller takes ownership of saving/collecting the entry itself, as done
+    for example in PurchaseOrder.receive_line_items()) is unchanged.
+
+    Nesting is not supported: a nested batch_tracking_entries() call reuses the outer batch,
+    and only the outermost call schedules a flush.
+
+    This mirrors plugin.base.event.events.batch_events() - see that docstring for the
+    reasoning behind the on-commit flush and the context-local (rather than parameter-based)
+    design, which allows existing single-item entrypoints (e.g. StockItem.stocktake()) to be
+    reused unmodified by both single-item and bulk callers.
+
+    Yields:
+        The current TrackingEntryBatch instance
+    """
+    if _tracking_batch.get() is not None:
+        # Already inside a batch - extend it, rather than creating a nested one
+        yield _tracking_batch.get()
+        return
+
+    batch = TrackingEntryBatch()
+    token = _tracking_batch.set(batch)
+
+    try:
+        yield batch
+    finally:
+        _tracking_batch.reset(token)
+        transaction.on_commit(batch.flush)
+
+
 class StockItemTracking(InvenTree.models.InvenTreeModel):
     """Stock tracking entry - used for tracking history of a particular StockItem.
 

src/backend/InvenTree/stock/serializers.py

@@ -31,6 +31,7 @@ import stock.status_codes
 from common.settings import get_global_setting
 from generic.states.fields import InvenTreeCustomStatusSerializerMixin
 from importer.registry import register_importer
+from InvenTree.fields import PrefetchedPrimaryKeyRelatedField
 from InvenTree.mixins import DataImportExportSerializerMixin
 from InvenTree.serializers import (
     CustomStatusSerializerMixin,
@@ -39,6 +40,8 @@ from InvenTree.serializers import (
     OptionalField,
     TreePathSerializer,
 )
+from InvenTree.tasks import batch_offload_tasks
+from plugin.base.event.events import batch_events
 from users.serializers import UserSerializer
 
 from .models import (
@@ -47,6 +50,7 @@ from .models import (
     StockItemTracking,
     StockLocation,
     StockLocationType,
+    batch_tracking_entries,
 )
 
 logger = structlog.get_logger('inventree')
@@ -809,6 +813,12 @@ class SerializeStockItemSerializer(serializers.Serializer):
             part=item.part,
         )
 
+        with (
+            transaction.atomic(),
+            batch_events(),
+            batch_tracking_entries(),
+            batch_offload_tasks(),
+        ):
             return (
                 item.serializeStock(
                     data['quantity'],
@@ -1853,7 +1863,8 @@ class StockAdjustmentItemSerializer(serializers.Serializer):
 
         super().__init__(*args, **kwargs)
 
-    pk = serializers.PrimaryKeyRelatedField(
+    pk = PrefetchedPrimaryKeyRelatedField(
+        cache_key='_stockitems',
         queryset=StockItem.objects.all(),
         many=False,
         allow_null=False,
@@ -1943,6 +1954,27 @@ class StockAdjustmentSerializer(serializers.Serializer):
         help_text=_('Stock transaction notes'),
     )
 
+    def to_internal_value(self, data):
+        """Bulk-fetch referenced StockItem objects before per-item validation.
+
+        Populates the '_stockitems' context cache that PrefetchedPrimaryKeyRelatedField
+        looks up against, avoiding one .get() query per item in the 'items' list.
+        """
+        pks = set()
+
+        for item in data.get('items', []):
+            try:
+                pks.add(int(item.get('pk')))
+            except (TypeError, ValueError):
+                pass
+
+        if pks:
+            self.context['_stockitems'] = {
+                obj.pk: obj for obj in StockItem.objects.filter(pk__in=pks)
+            }
+
+        return super().to_internal_value(data)
+
     def validate(self, data):
         """Make sure items are provided."""
         super().validate(data)
@@ -1990,7 +2022,12 @@ class StockCountSerializer(StockAdjustmentSerializer):
         notes = data.get('notes', '')
         location = data.get('location', None)
 
-        with transaction.atomic():
+        with (
+            transaction.atomic(),
+            batch_events(),
+            batch_tracking_entries(),
+            batch_offload_tasks(),
+        ):
             for item in items:
                 stock_item = item['pk']
                 quantity = item['quantity']
@@ -2018,7 +2055,12 @@ class StockAddSerializer(StockAdjustmentSerializer):
         data = self.validated_data
         notes = data.get('notes', '')
 
-        with transaction.atomic():
+        with (
+            transaction.atomic(),
+            batch_events(),
+            batch_tracking_entries(),
+            batch_offload_tasks(),
+        ):
             for item in data['items']:
                 stock_item = item['pk']
                 quantity = item['quantity']
@@ -2047,7 +2089,12 @@ class StockRemoveSerializer(StockAdjustmentSerializer):
         data = self.validated_data
         notes = data.get('notes', '')
 
-        with transaction.atomic():
+        with (
+            transaction.atomic(),
+            batch_events(),
+            batch_tracking_entries(),
+            batch_offload_tasks(),
+        ):
             for item in data['items']:
                 stock_item = item['pk']
                 quantity = item['quantity']
@@ -2104,7 +2151,12 @@ class StockTransferSerializer(StockAdjustmentSerializer):
         notes = data.get('notes', '')
         location = data['location']
 
-        with transaction.atomic():
+        with (
+            transaction.atomic(),
+            batch_events(),
+            batch_tracking_entries(),
+            batch_offload_tasks(),
+        ):
             for item in items:
                 # Required fields
                 stock_item = item['pk']

src/backend/InvenTree/stock/test_api.py

@@ -1554,7 +1554,12 @@ class StockItemTest(StockAPITestCase):
         """Test creation of a StockItem via the API."""
         # POST with an empty part reference
 
-        response = self.client.post(self.list_url, data={'quantity': 10, 'location': 1})
+        response = self.post(
+            self.list_url,
+            data={'quantity': 10, 'location': 1},
+            max_query_count=2250,
+            expected_code=400,
+        )
 
         self.assertContains(
             response,
@@ -1564,8 +1569,11 @@ class StockItemTest(StockAPITestCase):
 
         # POST with an invalid part reference
 
-        response = self.client.post(
+        response = self.post(
-            self.list_url, data={'quantity': 10, 'location': 1, 'part': 10000000}
+            self.list_url,
+            data={'quantity': 10, 'location': 1, 'part': 10000000},
+            max_query_count=2250,
+            expected_code=400,
         )
 
         self.assertContains(
@@ -1631,6 +1639,37 @@ class StockItemTest(StockAPITestCase):
         for new_item in new_items:
             self.assertIsNotNone(new_item.creation_date)
 
+    def test_bulk_serialize_benchmark(self):
+        """Benchmark: measure the number of DB queries required to serialize 100 stock items at once."""
+        InvenTreeSetting.set_setting('ENABLE_PLUGINS_EVENTS', True, change_user=None)
+
+        part = Part.objects.create(
+            name='Bulk serialize benchmark part',
+            description='Created for the stock serialize query-count benchmark',
+            trackable=True,
+        )
+
+        location = StockLocation.objects.create(
+            name='Bulk serialize benchmark location'
+        )
+
+        item = StockItem.objects.create(part=part, location=location, quantity=100)
+
+        url = reverse('api-stock-item-serialize', kwargs={'pk': item.pk})
+
+        data = {'quantity': 100, 'serial_numbers': '1-100', 'destination': location.pk}
+
+        with self.settings(
+            PLUGIN_TESTING_EVENTS=True, PLUGIN_TESTING_EVENTS_ASYNC=True
+        ):
+            # TODO: 2026-07-12 : Refactor this API call
+            response = self.post(
+                url, data, max_query_count=1300, benchmark=True, format='json'
+            )
+
+        self.assertEqual(response.status_code, 201)
+        self.assertEqual(len(response.data), 100)
+
     def test_stock_item_create_with_supplier_part(self):
         """Test creation of a StockItem via the API, including SupplierPart data."""
         # POST with non-existent supplier part
@@ -2922,6 +2961,140 @@ class StocktakeTest(StockAPITestCase):
             str(response.data['location']),
         )
 
+    def test_bulk_count_query_benchmark(self):
+        """Benchmark: measure the number of DB queries required to count 100 stock items at once."""
+        InvenTreeSetting.set_setting('ENABLE_PLUGINS_EVENTS', True, change_user=None)
+
+        part = Part.objects.create(
+            name='Bulk count benchmark part',
+            description='Created for the stock count query-count benchmark',
+        )
+
+        location = StockLocation.objects.create(name='Bulk count benchmark location')
+
+        items = [
+            StockItem.objects.create(part=part, location=location, quantity=idx + 1)
+            for idx in range(100)
+        ]
+
+        url = reverse('api-stock-count')
+
+        data = {
+            'items': [
+                {'pk': item.pk, 'quantity': idx + 100} for idx, item in enumerate(items)
+            ]
+        }
+
+        with self.settings(
+            PLUGIN_TESTING_EVENTS=True, PLUGIN_TESTING_EVENTS_ASYNC=True
+        ):
+            # TODO: 2026-07-12 : Refactor this API call
+            response = self.post(
+                url, data, max_query_count=2250, benchmark=True, format='json'
+            )
+
+        self.assertEqual(response.status_code, 201)
+        self.assertEqual(len(response.data['items']), 100)
+
+    def test_bulk_add_query_benchmark(self):
+        """Benchmark: measure the number of DB queries required to add stock to 100 items at once."""
+        InvenTreeSetting.set_setting('ENABLE_PLUGINS_EVENTS', True, change_user=None)
+
+        part = Part.objects.create(
+            name='Bulk add benchmark part',
+            description='Created for the stock add query-count benchmark',
+        )
+
+        location = StockLocation.objects.create(name='Bulk add benchmark location')
+
+        items = [
+            StockItem.objects.create(part=part, location=location, quantity=idx + 1)
+            for idx in range(100)
+        ]
+
+        url = reverse('api-stock-add')
+
+        data = {'items': [{'pk': item.pk, 'quantity': 5} for item in items]}
+
+        with self.settings(
+            PLUGIN_TESTING_EVENTS=True, PLUGIN_TESTING_EVENTS_ASYNC=True
+        ):
+            # TODO: 2026-07-12 : Refactor this API call
+            response = self.post(
+                url, data, max_query_count=2500, benchmark=True, format='json'
+            )
+
+        self.assertEqual(response.status_code, 201)
+        self.assertEqual(len(response.data['items']), 100)
+
+    def test_bulk_remove_query_benchmark(self):
+        """Benchmark: measure the number of DB queries required to remove stock from 100 items at once."""
+        InvenTreeSetting.set_setting('ENABLE_PLUGINS_EVENTS', True, change_user=None)
+
+        part = Part.objects.create(
+            name='Bulk remove benchmark part',
+            description='Created for the stock remove query-count benchmark',
+        )
+
+        location = StockLocation.objects.create(name='Bulk remove benchmark location')
+
+        items = [
+            StockItem.objects.create(part=part, location=location, quantity=idx + 100)
+            for idx in range(100)
+        ]
+
+        url = reverse('api-stock-remove')
+
+        data = {'items': [{'pk': item.pk, 'quantity': 5} for item in items]}
+
+        with self.settings(
+            PLUGIN_TESTING_EVENTS=True, PLUGIN_TESTING_EVENTS_ASYNC=True
+        ):
+            # TODO: 2026-07-12 : Refactor this API call
+            response = self.post(
+                url, data, max_query_count=2250, benchmark=True, format='json'
+            )
+
+        self.assertEqual(response.status_code, 201)
+        self.assertEqual(len(response.data['items']), 100)
+
+    def test_bulk_move_query_benchmark(self):
+        """Benchmark: measure the number of DB queries required to move 100 stock items at once."""
+        InvenTreeSetting.set_setting('ENABLE_PLUGINS_EVENTS', True, change_user=None)
+
+        part = Part.objects.create(
+            name='Bulk move benchmark part',
+            description='Created for the stock move query-count benchmark',
+        )
+
+        source = StockLocation.objects.create(name='Bulk move benchmark source')
+        destination = StockLocation.objects.create(
+            name='Bulk move benchmark destination'
+        )
+
+        items = [
+            StockItem.objects.create(part=part, location=source, quantity=idx + 1)
+            for idx in range(100)
+        ]
+
+        url = reverse('api-stock-transfer')
+
+        data = {
+            'items': [{'pk': item.pk, 'quantity': item.quantity} for item in items],
+            'location': destination.pk,
+        }
+
+        with self.settings(
+            PLUGIN_TESTING_EVENTS=True, PLUGIN_TESTING_EVENTS_ASYNC=True
+        ):
+            # TODO: 2026-07-12 : Refactor this API call
+            response = self.post(
+                url, data, max_query_count=1250, benchmark=True, format='json'
+            )
+
+        self.assertEqual(response.status_code, 201)
+        self.assertEqual(len(response.data['items']), 100)
+
 
 class StockTransferMergeTest(StockAPITestCase):
     """Tests for optional merge-on-transfer behavior."""
@@ -3024,6 +3197,9 @@ class StockTransferMergeTest(StockAPITestCase):
 
     def test_transfer_merge_does_not_copy_source_tracking(self):
         """Transfer merge keeps destination history and adds a single merge entry."""
+        # Track total number of tracking entries created
+        N_TRACKING_ENTRIES = StockItemTracking.objects.count()
+
         existing = StockItem.objects.create(
             part=self.part, location=self.dest, quantity=100
         )
@@ -3062,6 +3238,23 @@ class StockTransferMergeTest(StockAPITestCase):
         self.assertEqual(merge_entry.deltas['stockitem'], incoming_pk)
         self.assertEqual(merge_entry.deltas['location'], self.dest.pk)
 
+        self.assertEqual(StockItemTracking.objects.count(), N_TRACKING_ENTRIES + 4)
+
+        # Ensure tracking entries were bulk created in the correct order
+        entries = list(StockItemTracking.objects.order_by('-pk')[:4])[::-1]
+
+        for idx, tt in enumerate([
+            StockHistoryCode.CREATED,
+            StockHistoryCode.CREATED,
+            StockHistoryCode.STOCK_UPDATE,
+            StockHistoryCode.MERGED_STOCK_ITEMS,
+        ]):
+            self.assertEqual(
+                entries[idx].tracking_type,
+                tt,
+                f'Entry {idx} has unexpected tracking type {entries[idx].tracking_type}',
+            )
+
     def test_transfer_merge_partial_reuses_split_transfer_deltas(self):
         """Partial merge reuses split transfer deltas on the merge tracking entry."""
         existing = StockItem.objects.create(

src/backend/InvenTree/stock/tests.py

@@ -3,9 +3,11 @@
 import datetime
 
 from django.core.exceptions import ValidationError
+from django.db import transaction
 from django.db.models import Sum
 from django.test import override_settings
 
+from django_q.models import OrmQ
 from djmoney.money import Money
 
 from build.models import Build
@@ -14,6 +16,8 @@ from company.models import Company
 from InvenTree.unit_test import AdminTestCase, InvenTreeTestCase
 from order.models import SalesOrder
 from part.models import Part, PartTestTemplate
+from plugin.base.event.events import batch_events
+from stock.events import StockEvents
 from stock.status_codes import StockHistoryCode, StockStatus
 
 from .models import (
@@ -22,6 +26,7 @@ from .models import (
     StockItemTracking,
     StockLocation,
     StockLocationType,
+    batch_tracking_entries,
 )
 
 
@@ -386,6 +391,76 @@ class StockTest(StockTestBase):
         self.assertEqual(it.quantity, 100)
         self.assertEqual(it.status, StockStatus.DAMAGED.value)
 
+    def get_counted_events(self):
+        """Helper: return queued OrmQ tasks corresponding to a StockEvents.ITEM_COUNTED event.
+
+        stocktake() also fires an ITEM_QUANTITY_UPDATED event (via updateQuantity()) and may
+        offload a (deduped) low-stock notification task, so tests filter down to just the
+        ITEM_COUNTED entries they care about, rather than asserting the raw queue total.
+        """
+        return [
+            task
+            for task in OrmQ.objects.all().order_by('id')
+            if task.args() == (StockEvents.ITEM_COUNTED,)
+        ]
+
+    def test_stocktake_batch_events(self):
+        """Test that batch_events() collapses per-item stocktake() events into one bulk write.
+
+        StockItem.stocktake() always calls trigger_event() exactly as it does when called
+        standalone (see test_stocktake_events_outside_batch below) - a bulk caller wraps its
+        loop in batch_events() to collapse those N individual event offloads into a single
+        bulk_trigger_event() call, fired when the enclosing transaction commits.
+        """
+        InvenTreeSetting.set_setting('ENABLE_PLUGINS_EVENTS', True, change_user=None)
+
+        part = Part.objects.create(
+            name='Batch stocktake part', description='For batch stocktake testing'
+        )
+
+        items = [
+            StockItem.objects.create(part=part, location=self.home, quantity=idx + 1)
+            for idx in range(10)
+        ]
+
+        OrmQ.objects.all().delete()
+
+        with self.settings(
+            PLUGIN_TESTING_EVENTS=True, PLUGIN_TESTING_EVENTS_ASYNC=True
+        ):
+            # All 10 ITEM_COUNTED events should be queued via a single bulk write,
+            # fired when the transaction commits (captured here via captureOnCommitCallbacks)
+            with self.captureOnCommitCallbacks(execute=True):
+                with transaction.atomic(), batch_events():
+                    for idx, item in enumerate(items):
+                        item.stocktake(100 + idx, self.user, notes='Batch stocktake')
+
+        counted_tasks = self.get_counted_events()
+        self.assertEqual(len(counted_tasks), 10)
+
+        for idx, (task, item) in enumerate(zip(counted_tasks, items, strict=True)):
+            self.assertEqual(task.func(), 'plugin.base.event.events.register_event')
+            self.assertEqual(task.kwargs(), {'id': item.id, 'quantity': 100.0 + idx})
+
+    def test_stocktake_events_outside_batch(self):
+        """Test that stocktake() still fires events immediately when called outside batch_events()."""
+        InvenTreeSetting.set_setting('ENABLE_PLUGINS_EVENTS', True, change_user=None)
+
+        item = StockItem.objects.get(pk=2)
+
+        OrmQ.objects.all().delete()
+
+        with self.settings(
+            PLUGIN_TESTING_EVENTS=True, PLUGIN_TESTING_EVENTS_ASYNC=True
+        ):
+            item.stocktake(42, self.user, notes='Single stocktake')
+
+        # No batch_events() context was active, so the event was queued immediately -
+        # no transaction commit or captureOnCommitCallbacks is required to see it
+        counted_tasks = self.get_counted_events()
+        self.assertEqual(len(counted_tasks), 1)
+        self.assertEqual(counted_tasks[0].kwargs(), {'id': item.id, 'quantity': 42.0})
+
     def test_add_stock(self):
         """Test adding stock."""
         it = StockItem.objects.get(pk=2)
@@ -832,6 +907,130 @@ class StockTest(StockTestBase):
             self.assertEqual(item.purchase_price_currency, 'GBP')
 
 
+class TrackingEntryBatchTests(StockTestBase):
+    """Unit tests for the batch_tracking_entries() context manager."""
+
+    def test_entries_queued_and_flushed_on_commit(self):
+        """Tracking entries created inside batch_tracking_entries() are bulk-created on commit."""
+        item = StockItem.objects.get(pk=2)
+
+        n = StockItemTracking.objects.count()
+
+        with self.captureOnCommitCallbacks(execute=True):
+            with transaction.atomic(), batch_tracking_entries():
+                for idx in range(10):
+                    item.add_tracking_entry(
+                        StockHistoryCode.STOCK_COUNT,
+                        self.user,
+                        deltas={'quantity': idx},
+                        notes=f'Batch entry {idx}',
+                    )
+
+                # Nothing should be written yet - the batch only flushes on commit
+                self.assertEqual(StockItemTracking.objects.count(), n)
+
+        entries = StockItemTracking.objects.filter(item=item).order_by('id')[:10]
+        self.assertEqual(StockItemTracking.objects.count(), n + 10)
+
+        for idx, entry in enumerate(entries):
+            self.assertEqual(entry.tracking_type, StockHistoryCode.STOCK_COUNT)
+            self.assertEqual(entry.deltas.get('quantity'), idx)
+            self.assertEqual(entry.notes, f'Batch entry {idx}')
+
+    def test_entries_discarded_on_rollback(self):
+        """Tracking entries queued in a batch are discarded if the transaction rolls back."""
+        item = StockItem.objects.get(pk=2)
+
+        n = StockItemTracking.objects.count()
+
+        with self.captureOnCommitCallbacks(execute=True):
+            try:
+                with transaction.atomic(), batch_tracking_entries():
+                    item.add_tracking_entry(
+                        StockHistoryCode.STOCK_COUNT, self.user, deltas={'quantity': 1}
+                    )
+                    raise ValueError('boom')
+            except ValueError:
+                pass
+
+        self.assertEqual(StockItemTracking.objects.count(), n)
+
+    def test_entries_outside_batch_fire_immediately(self):
+        """Tracking entries created outside any batch are unaffected - saved immediately."""
+        item = StockItem.objects.get(pk=2)
+
+        n = StockItemTracking.objects.count()
+
+        item.add_tracking_entry(
+            StockHistoryCode.STOCK_COUNT, self.user, deltas={'quantity': 1}
+        )
+
+        # No transaction commit or captureOnCommitCallbacks needed - it was never queued
+        self.assertEqual(StockItemTracking.objects.count(), n + 1)
+
+    def test_nested_batch_share_one_flush(self):
+        """A nested batch_tracking_entries() call reuses the outer batch, rather than flushing twice."""
+        item = StockItem.objects.get(pk=2)
+
+        n = StockItemTracking.objects.count()
+
+        with self.captureOnCommitCallbacks(execute=True):
+            with transaction.atomic(), batch_tracking_entries():
+                item.add_tracking_entry(
+                    StockHistoryCode.STOCK_COUNT, self.user, deltas={'quantity': 1}
+                )
+                with batch_tracking_entries():
+                    item.add_tracking_entry(
+                        StockHistoryCode.STOCK_COUNT, self.user, deltas={'quantity': 2}
+                    )
+                self.assertEqual(StockItemTracking.objects.count(), n)
+
+        self.assertEqual(StockItemTracking.objects.count(), n + 2)
+
+    def test_commit_false_is_unaffected_by_batch(self):
+        """commit=False callers keep manual ownership of the entry, even inside a batch."""
+        item = StockItem.objects.get(pk=2)
+
+        n = StockItemTracking.objects.count()
+
+        with transaction.atomic(), batch_tracking_entries():
+            entry = item.add_tracking_entry(
+                StockHistoryCode.STOCK_COUNT,
+                self.user,
+                deltas={'quantity': 1},
+                commit=False,
+            )
+
+        # The entry was returned uncommitted - the batch never saw it, so it was never written
+        self.assertIsNotNone(entry)
+        self.assertIsNone(entry.pk)
+        self.assertEqual(StockItemTracking.objects.count(), n)
+
+    def test_stocktake_batch_tracking_entries(self):
+        """Integration test: batching stocktake() tracking entries via the StockCountSerializer path."""
+        part = Part.objects.create(
+            name='Batch tracking part', description='For batch tracking testing'
+        )
+
+        items = [
+            StockItem.objects.create(part=part, location=self.home, quantity=idx + 1)
+            for idx in range(10)
+        ]
+
+        n = StockItemTracking.objects.count()
+
+        with self.captureOnCommitCallbacks(execute=True):
+            with transaction.atomic(), batch_tracking_entries():
+                for idx, item in enumerate(items):
+                    item.stocktake(100 + idx, self.user, notes='Batch stocktake')
+
+        entries = StockItemTracking.objects.filter(
+            item__in=items, tracking_type=StockHistoryCode.STOCK_COUNT
+        ).order_by('id')
+        self.assertEqual(entries.count(), 10)
+        self.assertEqual(StockItemTracking.objects.count(), n + 10)
+
+
 class StockBarcodeTest(StockTestBase):
     """Run barcode tests for the stock app."""