From b42521f205079d30cda1c3474f5ef48098cfdd35 Mon Sep 17 00:00:00 2001 From: Oliver Date: Sun, 12 Jul 2026 15:58:09 +1000 Subject: [PATCH] [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 | 37 ++++ src/backend/InvenTree/InvenTree/tasks.py | 186 +++++++++++++++- src/backend/InvenTree/InvenTree/test_tasks.py | 173 +++++++++++++++ src/backend/InvenTree/InvenTree/unit_test.py | 25 ++- src/backend/InvenTree/order/serializers.py | 15 +- src/backend/InvenTree/order/test_api.py | 41 ++++ .../InvenTree/plugin/base/event/events.py | 131 +++++++++++- .../plugin/base/event/test_events.py | 134 ++++++++++++ src/backend/InvenTree/plugin/events.py | 15 +- src/backend/InvenTree/stock/models.py | 78 ++++++- src/backend/InvenTree/stock/serializers.py | 80 +++++-- src/backend/InvenTree/stock/test_api.py | 199 +++++++++++++++++- src/backend/InvenTree/stock/tests.py | 199 ++++++++++++++++++ 13 files changed, 1283 insertions(+), 30 deletions(-) create mode 100644 src/backend/InvenTree/plugin/base/event/test_events.py diff --git a/src/backend/InvenTree/InvenTree/fields.py b/src/backend/InvenTree/InvenTree/fields.py index 1445b83ecb..8f6d55bde5 100644 --- a/src/backend/InvenTree/InvenTree/fields.py +++ b/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.""" diff --git a/src/backend/InvenTree/InvenTree/tasks.py b/src/backend/InvenTree/InvenTree/tasks.py index e44520629b..0b868920b8 100644 --- a/src/backend/InvenTree/InvenTree/tasks.py +++ b/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. diff --git a/src/backend/InvenTree/InvenTree/test_tasks.py b/src/backend/InvenTree/InvenTree/test_tasks.py index 0638c51d9b..c059e79254 100644 --- a/src/backend/InvenTree/InvenTree/test_tasks.py +++ b/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) diff --git a/src/backend/InvenTree/InvenTree/unit_test.py b/src/backend/InvenTree/InvenTree/unit_test.py index bae1262297..572f8930cc 100644 --- a/src/backend/InvenTree/InvenTree/unit_test.py +++ b/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): diff --git a/src/backend/InvenTree/order/serializers.py b/src/backend/InvenTree/order/serializers.py index 04dc6e79f4..c69045a205 100644 --- a/src/backend/InvenTree/order/serializers.py +++ b/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,9 +1050,15 @@ class PurchaseOrderReceiveSerializer(serializers.Serializer): location = data.get('location', order.destination) try: - items = order.receive_line_items( - location, items, request.user if request else None - ) + with ( + transaction.atomic(), + batch_events(), + batch_tracking_entries(), + batch_offload_tasks(), + ): + items = order.receive_line_items( + location, items, request.user if request else None + ) except (ValidationError, DjangoValidationError) as exc: # Catch model errors and re-throw as DRF errors raise ValidationError(detail=serializers.as_serializer_error(exc)) diff --git a/src/backend/InvenTree/order/test_api.py b/src/backend/InvenTree/order/test_api.py index e72bd91de5..99a70e534f 100644 --- a/src/backend/InvenTree/order/test_api.py +++ b/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) diff --git a/src/backend/InvenTree/plugin/base/event/events.py b/src/backend/InvenTree/plugin/base/event/events.py index 1a77432804..c834e02ae4 100644 --- a/src/backend/InvenTree/plugin/base/event/events.py +++ b/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. diff --git a/src/backend/InvenTree/plugin/base/event/test_events.py b/src/backend/InvenTree/plugin/base/event/test_events.py new file mode 100644 index 0000000000..77c5c5f998 --- /dev/null +++ b/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) diff --git a/src/backend/InvenTree/plugin/events.py b/src/backend/InvenTree/plugin/events.py index c4f9d8006b..2097767440 100644 --- a/src/backend/InvenTree/plugin/events.py +++ b/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', +] diff --git a/src/backend/InvenTree/stock/models.py b/src/backend/InvenTree/stock/models.py index f033f56769..43534d57e9 100644 --- a/src/backend/InvenTree/stock/models.py +++ b/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,7 +2216,11 @@ class StockItem( ) if commit: - entry.save() + 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. diff --git a/src/backend/InvenTree/stock/serializers.py b/src/backend/InvenTree/stock/serializers.py index bae45c9d51..03764b8cf2 100644 --- a/src/backend/InvenTree/stock/serializers.py +++ b/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,16 +813,22 @@ class SerializeStockItemSerializer(serializers.Serializer): part=item.part, ) - return ( - item.serializeStock( - data['quantity'], - serials, - user=user, - notes=data.get('notes', ''), - location=data['destination'], + with ( + transaction.atomic(), + batch_events(), + batch_tracking_entries(), + batch_offload_tasks(), + ): + return ( + item.serializeStock( + data['quantity'], + serials, + user=user, + notes=data.get('notes', ''), + location=data['destination'], + ) + or [] ) - or [] - ) class InstallStockItemSerializer(serializers.Serializer): @@ -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'] diff --git a/src/backend/InvenTree/stock/test_api.py b/src/backend/InvenTree/stock/test_api.py index 19b6e082e6..8347a75b6f 100644 --- a/src/backend/InvenTree/stock/test_api.py +++ b/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( - self.list_url, data={'quantity': 10, 'location': 1, 'part': 10000000} + response = self.post( + 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( diff --git a/src/backend/InvenTree/stock/tests.py b/src/backend/InvenTree/stock/tests.py index c0715bda64..b0c6e828af 100644 --- a/src/backend/InvenTree/stock/tests.py +++ b/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."""