Coverage for src/backend/InvenTree/stock/api.py: 60%
624 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 17:47 +0000
1"""JSON API for the Stock app."""
3from collections import OrderedDict
4from datetime import timedelta
6from django.core.exceptions import ValidationError as DjangoValidationError
7from django.db import transaction
8from django.db.models import F, Q
9from django.urls import include, path
10from django.utils.translation import gettext_lazy as _
12import django_filters.rest_framework.filters as rest_filters
13from django_filters.rest_framework.filterset import FilterSet
14from drf_spectacular.types import OpenApiTypes
15from drf_spectacular.utils import extend_schema, extend_schema_field
16from rest_framework import status
17from rest_framework.generics import GenericAPIView
18from rest_framework.response import Response
19from rest_framework.serializers import ValidationError
21import common.filters
22import common.models
23import common.settings
24import InvenTree.helpers
25import InvenTree.permissions
26import stock.serializers as StockSerializers
27from build.models import Build
28from build.serializers import BuildSerializer
29from company.models import Company, ManufacturerPart, SupplierPart
30from company.serializers import CompanySerializer
31from data_exporter.mixins import DataExportViewMixin
32from generic.states.api import StatusView
33from InvenTree.api import (
34 BulkCreateMixin,
35 BulkUpdateMixin,
36 ListCreateDestroyAPIView,
37 TreeMixin,
38 meta_path,
39)
40from InvenTree.fields import InvenTreeOutputOption, OutputConfiguration
41from InvenTree.filters import (
42 SEARCH_ORDER_FILTER,
43 InvenTreeDateFilter,
44 NumberOrNullFilter,
45)
46from InvenTree.helpers import extract_serial_numbers, generateTestKey, str2bool
47from InvenTree.mixins import (
48 CreateAPI,
49 CustomRetrieveUpdateDestroyAPI,
50 ListAPI,
51 ListCreateAPI,
52 OutputOptionsMixin,
53 RetrieveAPI,
54 RetrieveUpdateDestroyAPI,
55 SerializerContextMixin,
56)
57from order.models import PurchaseOrder, ReturnOrder, SalesOrder, TransferOrder
58from order.serializers import (
59 PurchaseOrderSerializer,
60 ReturnOrderSerializer,
61 SalesOrderSerializer,
62 TransferOrderSerializer,
63)
64from part.models import BomItem, Part, PartCategory
65from part.serializers import PartBriefSerializer
66from stock.generators import generate_batch_code, generate_serial_number
67from stock.models import (
68 StockItem,
69 StockItemTestResult,
70 StockItemTracking,
71 StockLocation,
72 StockLocationType,
73)
74from stock.status_codes import StockHistoryCode, StockStatus
77class GenerateBatchCode(GenericAPIView):
78 """API endpoint for generating batch codes."""
80 permission_classes = [InvenTree.permissions.IsAuthenticatedOrReadScope]
81 serializer_class = StockSerializers.GenerateBatchCodeSerializer
83 def post(self, request, *args, **kwargs):
84 """Generate a new batch code."""
85 serializer = self.get_serializer(data=request.data)
86 serializer.is_valid(raise_exception=True)
88 data = {'batch_code': generate_batch_code(**serializer.validated_data)}
90 return Response(data, status=status.HTTP_201_CREATED)
93class GenerateSerialNumber(GenericAPIView):
94 """API endpoint for generating serial numbers."""
96 permission_classes = [InvenTree.permissions.IsAuthenticatedOrReadScope]
97 serializer_class = StockSerializers.GenerateSerialNumberSerializer
99 def post(self, request, *args, **kwargs):
100 """Generate a new serial number."""
101 serializer = self.get_serializer(data=request.data)
102 serializer.is_valid(raise_exception=True)
104 data = {'serial_number': generate_serial_number(**serializer.validated_data)}
106 return Response(data, status=status.HTTP_201_CREATED)
109class StockItemContextMixin:
110 """Mixin class for adding StockItem object to serializer context."""
112 role_required = 'stock.change'
114 queryset = StockItem.objects.none()
116 def get_serializer_context(self):
117 """Extend serializer context."""
118 context = super().get_serializer_context()
119 context['request'] = self.request
121 try:
122 context['item'] = StockItem.objects.get(pk=self.kwargs.get('pk', None))
123 except Exception: # pragma: no cover
124 pass
126 return context
129@extend_schema(responses={201: StockSerializers.StockItemSerializer(many=True)})
130class StockItemSerialize(StockItemContextMixin, CreateAPI):
131 """API endpoint for serializing a stock item."""
133 serializer_class = StockSerializers.SerializeStockItemSerializer
134 pagination_class = None
136 def create(self, request, *args, **kwargs):
137 """Serialize the provided StockItem."""
138 serializer = self.get_serializer(data=request.data)
139 serializer.is_valid(raise_exception=True)
141 # Perform the actual serialization step
142 items = serializer.save()
144 queryset = StockSerializers.StockItemSerializer.annotate_queryset(items)
146 response = StockSerializers.StockItemSerializer(
147 queryset, many=True, context=self.get_serializer_context()
148 )
150 return Response(response.data, status=status.HTTP_201_CREATED)
153class StockItemInstall(StockItemContextMixin, CreateAPI):
154 """API endpoint for installing a particular stock item into this stock item.
156 - stock_item.part must be in the BOM for this part
157 - stock_item must currently be "in stock"
158 - stock_item must be serialized (and not belong to another item)
159 """
161 serializer_class = StockSerializers.InstallStockItemSerializer
164class StockItemUninstall(StockItemContextMixin, CreateAPI):
165 """API endpoint for removing (uninstalling) items from this item."""
167 serializer_class = StockSerializers.UninstallStockItemSerializer
170class StockItemConvert(StockItemContextMixin, CreateAPI):
171 """API endpoint for converting a stock item to a variant part."""
173 serializer_class = StockSerializers.ConvertStockItemSerializer
176class StockAdjustView(CreateAPI):
177 """A generic class for handling stocktake actions.
179 Subclasses exist for:
181 - StockCount: count stock items
182 - StockAdd: add stock items
183 - StockRemove: remove stock items
184 - StockTransfer: transfer stock items
185 """
187 queryset = StockItem.objects.none()
189 def get_serializer_context(self):
190 """Extend serializer context."""
191 context = super().get_serializer_context()
192 context['request'] = self.request
194 return context
197class StockChangeStatus(StockAdjustView):
198 """API endpoint to change the status code of multiple StockItem objects."""
200 serializer_class = StockSerializers.StockChangeStatusSerializer
203class StockCount(StockAdjustView):
204 """Endpoint for counting stock (performing a stocktake)."""
206 serializer_class = StockSerializers.StockCountSerializer
209class StockAdd(StockAdjustView):
210 """Endpoint for adding a quantity of stock to an existing StockItem."""
212 serializer_class = StockSerializers.StockAddSerializer
215class StockRemove(StockAdjustView):
216 """Endpoint for removing a quantity of stock from an existing StockItem."""
218 serializer_class = StockSerializers.StockRemoveSerializer
221class StockTransfer(StockAdjustView):
222 """API endpoint for performing stock movements."""
224 serializer_class = StockSerializers.StockTransferSerializer
227class StockReturn(StockAdjustView):
228 """API endpoint for returning items into stock.
230 This API endpoint is for items that are initially considered "not in stock",
231 and the user wants to return them to stock, marking them as
232 "available" for further consumption or sale.
233 """
235 serializer_class = StockSerializers.StockReturnSerializer
238class StockAssign(CreateAPI):
239 """API endpoint for assigning stock to a particular customer."""
241 queryset = StockItem.objects.all()
242 serializer_class = StockSerializers.StockAssignmentSerializer
244 def get_serializer_context(self):
245 """Extend serializer context."""
246 ctx = super().get_serializer_context()
247 ctx['request'] = self.request
249 return ctx
252class StockMerge(CreateAPI):
253 """API endpoint for merging multiple stock items."""
255 queryset = StockItem.objects.none()
256 serializer_class = StockSerializers.StockMergeSerializer
258 def get_serializer_context(self):
259 """Extend serializer context."""
260 ctx = super().get_serializer_context()
261 ctx['request'] = self.request
262 return ctx
265class StockLocationFilter(FilterSet):
266 """Base class for custom API filters for the StockLocation endpoint."""
268 class Meta:
269 """Meta class options for this filterset."""
271 model = StockLocation
272 fields = ['name', 'structural', 'external']
274 location_type = rest_filters.ModelChoiceFilter(
275 queryset=StockLocationType.objects.all(), field_name='location_type'
276 )
278 has_location_type = rest_filters.BooleanFilter(
279 label='has_location_type', method='filter_has_location_type'
280 )
282 def filter_has_location_type(self, queryset, name, value):
283 """Filter by whether or not the location has a location type."""
284 if str2bool(value):
285 return queryset.exclude(location_type=None)
286 return queryset.filter(location_type=None)
288 depth = rest_filters.NumberFilter(
289 label=_('Depth'), method='filter_depth', help_text=_('Filter by location depth')
290 )
292 def filter_depth(self, queryset, name, value):
293 """Filter by the "depth" of the StockLocation.
295 - This filter is used to limit the depth of the location tree
296 - If the "parent" filter is also provided, the depth is calculated from the parent location
297 """
298 parent = self.data.get('parent', None)
300 # Only filter if the parent filter is *not* provided
301 if not parent: 301 ↛ 304line 301 didn't jump to line 304 because the condition on line 301 was always true
302 queryset = queryset.filter(level__lte=value)
304 return queryset
306 top_level = rest_filters.BooleanFilter(
307 label=_('Top Level'),
308 method='filter_top_level',
309 help_text=_('Filter by top-level locations'),
310 )
312 def filter_top_level(self, queryset, name, value):
313 """Filter by top-level locations."""
314 cascade = str2bool(self.data.get('cascade', False))
316 if value and not cascade:
317 return queryset.filter(parent=None)
319 return queryset
321 cascade = rest_filters.BooleanFilter(
322 label=_('Cascade'),
323 method='filter_cascade',
324 help_text=_('Include sub-locations in filtered results'),
325 )
327 def filter_cascade(self, queryset, name, value):
328 """Filter by whether to include sub-locations in the filtered results.
330 Note: If the "parent" filter is provided, we offload the logic to that method.
331 """
332 parent = self.data.get('parent', None)
333 top_level = str2bool(self.data.get('top_level', None))
335 # If the parent is *not* provided, update the results based on the "cascade" value
336 if (not parent or top_level) and not value:
337 # If "cascade" is False, only return top-level location
338 queryset = queryset.filter(parent=None)
340 return queryset
342 parent = rest_filters.ModelChoiceFilter(
343 queryset=StockLocation.objects.all(),
344 method='filter_parent',
345 label=_('Parent Location'),
346 help_text=_('Filter by parent location'),
347 )
349 def filter_parent(self, queryset, name, value):
350 """Filter by parent location.
352 Note that the filtering behavior here varies,
353 depending on whether the 'cascade' value is set.
355 So, we have to check the "cascade" value here.
356 """
357 parent = value
358 depth = self.data.get('depth', None)
359 cascade = str2bool(self.data.get('cascade', False))
361 if cascade:
362 # Return recursive sub-locations
363 queryset = queryset.filter(
364 parent__in=parent.get_descendants(include_self=True)
365 )
366 else:
367 # Return only direct children
368 queryset = queryset.filter(parent=parent)
370 if depth is not None:
371 # Filter by depth from parent
372 depth = int(depth)
373 queryset = queryset.filter(level__lte=parent.level + depth)
375 return queryset
377 tags = common.filters.TagsFilter(label=_('Tags'))
380class StockLocationMixin(SerializerContextMixin):
381 """Mixin class for StockLocation API endpoints."""
383 queryset = StockLocation.objects.all()
384 serializer_class = StockSerializers.LocationSerializer
386 def get_queryset(self, *args, **kwargs):
387 """Return annotated queryset for the StockLocationList endpoint."""
388 queryset = super().get_queryset(*args, **kwargs)
389 queryset = StockSerializers.LocationSerializer.annotate_queryset(queryset)
390 return queryset
393class StockLocationOutputOptions(OutputConfiguration):
394 """Output options for StockLocation serializers."""
396 OPTIONS = [
397 InvenTreeOutputOption(
398 description='Include detailed information about the BOM item linked to this build line.',
399 flag='path_detail',
400 )
401 ]
404class StockLocationList(
405 DataExportViewMixin,
406 BulkUpdateMixin,
407 StockLocationMixin,
408 OutputOptionsMixin,
409 ListCreateAPI,
410):
411 """API endpoint for list view of StockLocation objects.
413 - GET: Return list of StockLocation objects
414 - POST: Create a new StockLocation
415 """
417 filterset_class = StockLocationFilter
418 filter_backends = SEARCH_ORDER_FILTER
419 output_options = StockLocationOutputOptions
421 search_fields = ['name', 'description', 'pathstring', 'tags__name', 'tags__slug']
423 ordering_fields = ['name', 'pathstring', 'items', 'level', 'tree_id', 'lft']
425 ordering = ['tree_id', 'lft', 'name']
428class StockLocationDetail(
429 StockLocationMixin, OutputOptionsMixin, CustomRetrieveUpdateDestroyAPI
430):
431 """API endpoint for detail view of StockLocation object."""
433 output_options = StockLocationOutputOptions
435 def destroy(self, request, *args, **kwargs):
436 """Delete a Stock location instance via the API."""
437 serializer = StockSerializers.LocationDeleteSerializer(data=request.data)
438 serializer.is_valid(raise_exception=True)
440 delete_stock_items = InvenTree.helpers.str2bool(
441 serializer.validated_data.get('delete_stock_items', False)
442 )
444 delete_sub_locations = InvenTree.helpers.str2bool(
445 serializer.validated_data.get('delete_sub_locations', False)
446 )
448 return super().destroy(
449 request,
450 *args,
451 **{
452 **kwargs,
453 'delete_sub_locations': delete_sub_locations,
454 'delete_stock_items': delete_stock_items,
455 },
456 )
459class LocationTreeFilter(FilterSet):
460 """Custom filterset class for the StockLocationTree endpoint."""
462 class Meta:
463 """Metaclass options for this filterset."""
465 model = StockLocation
466 fields = ['parent', 'tree_id', 'level']
468 max_level = rest_filters.NumberFilter(
469 label=_('Max Level'),
470 method='filter_max_level',
471 help_text=_('Limit the depth of the category tree'),
472 )
474 def filter_max_level(self, queryset, name, value):
475 """Filter by the maximum depth of the category tree."""
476 return queryset.filter(level__lte=value)
479class StockLocationTree(TreeMixin, ListAPI):
480 """API endpoint for accessing a list of StockLocation objects, ready for rendering as a tree."""
482 model_class = StockLocation
483 queryset = StockLocation.objects.all()
484 serializer_class = StockSerializers.LocationTreeSerializer
485 filterset_class = LocationTreeFilter
487 def get_queryset(self, *args, **kwargs):
488 """Return annotated queryset for the StockLocationTree endpoint."""
489 queryset = super().get_queryset(*args, **kwargs)
490 queryset = StockSerializers.LocationTreeSerializer.annotate_queryset(queryset)
491 return queryset
494class StockLocationTypeList(ListCreateAPI):
495 """API endpoint for a list of StockLocationType objects.
497 - GET: Return a list of all StockLocationType objects
498 - POST: Create a StockLocationType
499 """
501 queryset = StockLocationType.objects.all()
502 serializer_class = StockSerializers.StockLocationTypeSerializer
504 filter_backends = SEARCH_ORDER_FILTER
506 ordering_fields = ['name', 'location_count', 'icon']
508 ordering = ['-location_count']
510 search_fields = ['name']
512 def get_queryset(self):
513 """Override the queryset method to include location count."""
514 queryset = super().get_queryset()
515 queryset = StockSerializers.StockLocationTypeSerializer.annotate_queryset(
516 queryset
517 )
519 return queryset
522class StockLocationTypeDetail(RetrieveUpdateDestroyAPI):
523 """API detail endpoint for a StockLocationType object.
525 - GET: return a single StockLocationType
526 - PUT: update a StockLocationType
527 - PATCH: partial update a StockLocationType
528 - DELETE: delete a StockLocationType
529 """
531 queryset = StockLocationType.objects.all()
532 serializer_class = StockSerializers.StockLocationTypeSerializer
534 def get_queryset(self):
535 """Override the queryset method to include location count."""
536 queryset = super().get_queryset()
537 queryset = StockSerializers.StockLocationTypeSerializer.annotate_queryset(
538 queryset
539 )
541 return queryset
544class StockFilter(FilterSet):
545 """FilterSet for StockItem LIST API."""
547 class Meta:
548 """Metaclass options for this filterset."""
550 model = StockItem
552 # Simple filter filters
553 fields = [
554 'supplier_part',
555 'belongs_to',
556 'build',
557 'customer',
558 'consumed_by',
559 'sales_order',
560 'purchase_order',
561 'tags__name',
562 'tags__slug',
563 ]
565 # Relationship filters
566 manufacturer = rest_filters.ModelChoiceFilter(
567 label='Manufacturer',
568 queryset=Company.objects.all(),
569 method='filter_manufacturer',
570 )
572 @extend_schema_field(OpenApiTypes.INT)
573 def filter_manufacturer(self, queryset, name, company):
574 """Filter by manufacturer."""
575 return queryset.filter(
576 Q(supplier_part__manufacturer_part__manufacturer__is_manufacturer=True)
577 & Q(supplier_part__manufacturer_part__manufacturer=company)
578 )
580 manufacturer_part = rest_filters.ModelChoiceFilter(
581 label=_('Manufacturer Part'),
582 queryset=ManufacturerPart.objects.all(),
583 field_name='supplier_part__manufacturer_part',
584 )
586 supplier = rest_filters.ModelChoiceFilter(
587 label=_('Supplier'),
588 queryset=Company.objects.filter(is_supplier=True),
589 field_name='supplier_part__supplier',
590 )
592 include_variants = rest_filters.BooleanFilter(
593 label=_('Include Variants'), method='filter_include_variants'
594 )
596 def filter_include_variants(self, queryset, name, value):
597 """Filter by whether or not to include variants of the selected part.
599 Note:
600 - This filter does nothing by itself, and requires the 'part' filter to be set.
601 - Refer to the 'filter_part' method for more information.
602 """
603 return queryset
605 part = rest_filters.ModelChoiceFilter(
606 label=_('Part'), queryset=Part.objects.all(), method='filter_part'
607 )
609 def filter_part(self, queryset, name, part):
610 """Filter StockItem list by provided Part instance.
612 Note:
613 - If "part" is a variant, include all variants of the selected part
614 - Otherwise, filter by the selected part
615 """
616 include_variants = str2bool(self.data.get('include_variants', True))
618 if include_variants:
619 return queryset.filter(part__in=part.get_descendants(include_self=True))
620 else:
621 return queryset.filter(part=part)
623 # Part name filters
624 name = rest_filters.CharFilter(
625 label=_('Part name (case insensitive)'),
626 field_name='part__name',
627 lookup_expr='iexact',
628 )
630 name_contains = rest_filters.CharFilter(
631 label=_('Part name contains (case insensitive)'),
632 field_name='part__name',
633 lookup_expr='icontains',
634 )
636 name_regex = rest_filters.CharFilter(
637 label=_('Part name (regex)'), field_name='part__name', lookup_expr='iregex'
638 )
640 # Part IPN filters
641 IPN = rest_filters.CharFilter(
642 label=_('Part IPN (case insensitive)'),
643 field_name='part__IPN',
644 lookup_expr='iexact',
645 )
647 IPN_contains = rest_filters.CharFilter(
648 label=_('Part IPN contains (case insensitive)'),
649 field_name='part__IPN',
650 lookup_expr='icontains',
651 )
653 IPN_regex = rest_filters.CharFilter(
654 label=_('Part IPN (regex)'), field_name='part__IPN', lookup_expr='iregex'
655 )
657 # Part attribute filters
658 assembly = rest_filters.BooleanFilter(
659 label=_('Assembly'), field_name='part__assembly'
660 )
662 active = rest_filters.BooleanFilter(label=_('Active'), field_name='part__active')
663 salable = rest_filters.BooleanFilter(label=_('Salable'), field_name='part__salable')
665 min_stock = rest_filters.NumberFilter(
666 label=_('Minimum stock'), field_name='quantity', lookup_expr='gte'
667 )
669 max_stock = rest_filters.NumberFilter(
670 label=_('Maximum stock'), field_name='quantity', lookup_expr='lte'
671 )
673 status = rest_filters.NumberFilter(label=_('Status Code'), method='filter_status')
675 def filter_status(self, queryset, name, value):
676 """Filter by integer status code.
678 Note: Also account for the possibility of a custom status code.
679 """
680 q1 = Q(status=value, status_custom_key__isnull=True)
681 q2 = Q(status_custom_key=value)
683 return queryset.filter(q1 | q2).distinct()
685 allocated = rest_filters.BooleanFilter(
686 label='Is Allocated', method='filter_allocated'
687 )
689 def filter_allocated(self, queryset, name, value):
690 """Filter by whether or not the stock item is 'allocated'."""
691 if str2bool(value):
692 # Filter StockItem with either build allocations or transfer order allocations or sales order allocations
693 return queryset.filter(
694 Q(sales_order_allocations__isnull=False)
695 | Q(transfer_order_allocations__isnull=False)
696 | Q(allocations__isnull=False)
697 ).distinct()
698 # Filter StockItem without build allocations or transfer order allocations or sales order allocations
699 return queryset.filter(
700 Q(sales_order_allocations__isnull=True)
701 & Q(transfer_order_allocations__isnull=True)
702 & Q(allocations__isnull=True)
703 )
705 expired = rest_filters.BooleanFilter(label='Expired', method='filter_expired')
707 def filter_expired(self, queryset, name, value):
708 """Filter by whether or not the stock item has expired."""
709 if not common.settings.stock_expiry_enabled():
710 return queryset
712 if str2bool(value):
713 return queryset.filter(StockItem.get_expired_filter())
714 return queryset.exclude(StockItem.get_expired_filter())
716 external = rest_filters.BooleanFilter(
717 label=_('External Location'), method='filter_external'
718 )
720 def filter_external(self, queryset, name, value):
721 """Filter by whether or not the stock item is located in an external location."""
722 if str2bool(value):
723 return queryset.filter(location__external=True)
724 return queryset.exclude(location__external=True)
726 in_stock = rest_filters.BooleanFilter(label='In Stock', method='filter_in_stock')
728 def filter_in_stock(self, queryset, name, value):
729 """Filter by if item is in stock."""
730 if str2bool(value): 730 ↛ 732line 730 didn't jump to line 732 because the condition on line 730 was always true
731 return queryset.filter(StockItem.IN_STOCK_FILTER)
732 return queryset.exclude(StockItem.IN_STOCK_FILTER)
734 available = rest_filters.BooleanFilter(label='Available', method='filter_available')
736 def filter_available(self, queryset, name, value):
737 """Filter by whether the StockItem is "available" or not.
739 Here, "available" means that the allocated quantity is less than the total quantity
740 """
741 if str2bool(value):
742 # The 'quantity' field is greater than the calculated 'allocated' field
743 # Note that the item must also be "in stock"
744 return queryset.filter(StockItem.IN_STOCK_FILTER).filter(
745 Q(quantity__gt=F('allocated'))
746 )
747 # The 'quantity' field is less than (or equal to) the calculated 'allocated' field
748 return queryset.filter(Q(quantity__lte=F('allocated')))
750 batch = rest_filters.CharFilter(
751 label='Batch code filter (case insensitive)', lookup_expr='iexact'
752 )
754 batch_regex = rest_filters.CharFilter(
755 label='Batch code filter (regex)', field_name='batch', lookup_expr='iregex'
756 )
758 is_building = rest_filters.BooleanFilter(label='In production')
760 # Serial number filtering
761 serial_gte = rest_filters.NumberFilter(
762 label='Serial number GTE', field_name='serial_int', lookup_expr='gte'
763 )
764 serial_lte = rest_filters.NumberFilter(
765 label='Serial number LTE', field_name='serial_int', lookup_expr='lte'
766 )
768 serial = rest_filters.CharFilter(
769 label='Serial number', field_name='serial', lookup_expr='exact'
770 )
772 serialized = rest_filters.BooleanFilter(
773 label='Has serial number', method='filter_serialized'
774 )
776 def filter_serialized(self, queryset, name, value):
777 """Filter by whether the StockItem has a serial number (or not)."""
778 q = Q(serial=None) | Q(serial='')
780 if str2bool(value):
781 return queryset.exclude(q)
783 return queryset.filter(q).distinct()
785 has_batch = rest_filters.BooleanFilter(
786 label='Has batch code', method='filter_has_batch'
787 )
789 def filter_has_batch(self, queryset, name, value):
790 """Filter by whether the StockItem has a batch code (or not)."""
791 q = Q(batch=None) | Q(batch='')
793 if str2bool(value):
794 return queryset.exclude(q)
796 return queryset.filter(q).distinct()
798 tracked = rest_filters.BooleanFilter(label='Tracked', method='filter_tracked')
800 def filter_tracked(self, queryset, name, value):
801 """Filter by whether this stock item is *tracked*.
803 Meaning either:
804 - It has a serial number
805 - It has a batch code
806 """
807 q_batch = Q(batch=None) | Q(batch='')
808 q_serial = Q(serial=None) | Q(serial='')
810 if str2bool(value):
811 return queryset.exclude(q_batch & q_serial)
813 return queryset.filter(q_batch).filter(q_serial).distinct()
815 consumed = rest_filters.BooleanFilter(
816 label=_('Consumed by Build Order'), method='filter_consumed'
817 )
819 def filter_consumed(self, queryset, name, value):
820 """Filter by whether the stock item has been consumed by a build order."""
821 if str2bool(value):
822 return queryset.filter(consumed_by__isnull=False)
823 return queryset.filter(consumed_by__isnull=True)
825 installed = rest_filters.BooleanFilter(
826 label=_('Installed in other stock item'), method='filter_installed'
827 )
829 def filter_installed(self, queryset, name, value):
830 """Filter stock items by "belongs_to" field being empty."""
831 if str2bool(value):
832 return queryset.exclude(belongs_to=None)
833 return queryset.filter(belongs_to=None)
835 has_installed_items = rest_filters.BooleanFilter(
836 label='Has installed items', method='filter_has_installed'
837 )
839 def filter_has_installed(self, queryset, name, value):
840 """Filter stock items by "belongs_to" field being empty."""
841 if str2bool(value):
842 return queryset.filter(installed_items__gt=0)
843 return queryset.filter(installed_items=0)
845 has_child_items = rest_filters.BooleanFilter(
846 label='Has child items', method='filter_has_child_items'
847 )
849 def filter_has_child_items(self, queryset, name, value):
850 """Filter stock items by "belongs_to" field being empty."""
851 if str2bool(value):
852 return queryset.filter(child_items__gt=0)
853 return queryset.filter(child_items=0)
855 sent_to_customer = rest_filters.BooleanFilter(
856 label='Sent to customer', method='filter_sent_to_customer'
857 )
859 def filter_sent_to_customer(self, queryset, name, value):
860 """Filter by sent to customer."""
861 if str2bool(value):
862 return queryset.exclude(customer=None)
863 return queryset.filter(customer=None)
865 depleted = rest_filters.BooleanFilter(label='Depleted', method='filter_depleted')
867 def filter_depleted(self, queryset, name, value):
868 """Filter by depleted items."""
869 if str2bool(value):
870 return queryset.filter(quantity__lte=0)
871 return queryset.exclude(quantity__lte=0)
873 has_purchase_price = rest_filters.BooleanFilter(
874 label='Has purchase price', method='filter_has_purchase_price'
875 )
877 def filter_has_purchase_price(self, queryset, name, value):
878 """Filter by having a purchase price."""
879 if str2bool(value):
880 return queryset.exclude(purchase_price=None)
881 return queryset.filter(purchase_price=None)
883 ancestor = rest_filters.ModelChoiceFilter(
884 label='Ancestor', queryset=StockItem.objects.all(), method='filter_ancestor'
885 )
887 @extend_schema_field(OpenApiTypes.INT)
888 def filter_ancestor(self, queryset, name, ancestor):
889 """Filter based on ancestor stock item."""
890 return queryset.filter(parent__in=ancestor.get_descendants(include_self=True))
892 category = rest_filters.ModelChoiceFilter(
893 label=_('Category'),
894 queryset=PartCategory.objects.all(),
895 method='filter_category',
896 )
898 @extend_schema_field(OpenApiTypes.INT)
899 def filter_category(self, queryset, name, category):
900 """Filter based on part category."""
901 child_categories = category.get_descendants(include_self=True)
903 return queryset.filter(part__category__in=child_categories)
905 bom_item = rest_filters.ModelChoiceFilter(
906 label=_('BOM Item'), queryset=BomItem.objects.all(), method='filter_bom_item'
907 )
909 @extend_schema_field(OpenApiTypes.INT)
910 def filter_bom_item(self, queryset, name, bom_item):
911 """Filter based on BOM item."""
912 return queryset.filter(bom_item.get_stock_filter())
914 part_tree = rest_filters.ModelChoiceFilter(
915 label=_('Part Tree'), queryset=Part.objects.all(), method='filter_part_tree'
916 )
918 @extend_schema_field(OpenApiTypes.INT)
919 def filter_part_tree(self, queryset, name, part_tree):
920 """Filter based on part tree."""
921 return queryset.filter(part__tree_id=part_tree.tree_id)
923 company = rest_filters.ModelChoiceFilter(
924 label=_('Company'), queryset=Company.objects.all(), method='filter_company'
925 )
927 @extend_schema_field(OpenApiTypes.INT)
928 def filter_company(self, queryset, name, company):
929 """Filter by company (either manufacturer or supplier)."""
930 return queryset.filter(
931 Q(supplier_part__supplier=company)
932 | Q(supplier_part__manufacturer_part__manufacturer=company)
933 ).distinct()
935 created_before = InvenTreeDateFilter(
936 label=_('Created before'), field_name='creation_date', lookup_expr='lt'
937 )
939 created_after = InvenTreeDateFilter(
940 label=_('Created after'), field_name='creation_date', lookup_expr='gt'
941 )
943 updated_before = InvenTreeDateFilter(
944 label=_('Updated before'), field_name='updated', lookup_expr='lt'
945 )
947 updated_after = InvenTreeDateFilter(
948 label=_('Updated after'), field_name='updated', lookup_expr='gt'
949 )
951 stocktake_before = InvenTreeDateFilter(
952 label=_('Stocktake Before'), field_name='stocktake_date', lookup_expr='lt'
953 )
955 stocktake_after = InvenTreeDateFilter(
956 label=_('Stocktake After'), field_name='stocktake_date', lookup_expr='gt'
957 )
959 has_stocktake = rest_filters.BooleanFilter(
960 label=_('Has Stocktake Date'), method='filter_has_stocktake'
961 )
963 def filter_has_stocktake(self, queryset, name, value):
964 """Filter by whether or not the StockItem has a stocktake date."""
965 if str2bool(value):
966 return queryset.exclude(stocktake_date=None)
967 return queryset.filter(stocktake_date=None)
969 # Stock "expiry" filters
970 expiry_before = InvenTreeDateFilter(
971 label=_('Expiry date before'), field_name='expiry_date', lookup_expr='lt'
972 )
974 expiry_after = InvenTreeDateFilter(
975 label=_('Expiry date after'), field_name='expiry_date', lookup_expr='gt'
976 )
978 stale = rest_filters.BooleanFilter(label=_('Stale'), method='filter_stale')
980 def filter_stale(self, queryset, name, value):
981 """Filter by stale stock items."""
982 stale_days = common.models.InvenTreeSetting.get_setting('STOCK_STALE_DAYS')
984 if stale_days <= 0:
985 # No filtering, does not make sense
986 return queryset
988 stale_date = InvenTree.helpers.current_date() + timedelta(days=stale_days)
989 stale_filter = (
990 StockItem.IN_STOCK_FILTER
991 & ~Q(expiry_date=None)
992 & Q(expiry_date__lt=stale_date)
993 )
995 if str2bool(value):
996 return queryset.filter(stale_filter)
997 else:
998 return queryset.exclude(stale_filter)
1000 exclude_tree = rest_filters.NumberFilter(
1001 method='filter_exclude_tree',
1002 label=_('Exclude Tree'),
1003 help_text=_(
1004 'Provide a StockItem PK to exclude that item and all its descendants'
1005 ),
1006 )
1008 def filter_exclude_tree(self, queryset, name, value):
1009 """Exclude a StockItem and all of its descendants from the queryset."""
1010 try:
1011 root = StockItem.objects.get(pk=value)
1012 pks_to_exclude = [
1013 item.pk for item in root.get_descendants(include_self=True)
1014 ]
1015 return queryset.exclude(pk__in=pks_to_exclude)
1016 except (ValueError, StockItem.DoesNotExist):
1017 # If the value is invalid or the object doesn't exist, do nothing.
1018 return queryset
1020 cascade = rest_filters.BooleanFilter(
1021 method='filter_cascade',
1022 label=_('Cascade Locations'),
1023 help_text=_('If true, include items in child locations of the given location'),
1024 )
1026 location = NumberOrNullFilter(
1027 method='filter_location',
1028 label=_('Location'),
1029 help_text=_("Filter by numeric Location ID or the literal 'null'"),
1030 )
1032 def filter_cascade(self, queryset, name, value):
1033 """Dummy filter method for 'cascade'.
1035 - Ensures 'cascade' appears in API documentation
1036 - Does NOT actually filter the queryset directly
1037 """
1038 return queryset
1040 def filter_location(self, queryset, name, value):
1041 """Filter for location that also applies cascade logic."""
1042 cascade = str2bool(self.data.get('cascade', True))
1044 if value == 'null':
1045 if not cascade:
1046 return queryset.filter(location=None)
1047 return queryset
1049 if not cascade:
1050 return queryset.filter(location=value)
1052 try:
1053 loc_obj = StockLocation.objects.get(pk=value)
1054 except StockLocation.DoesNotExist:
1055 return queryset
1057 children = loc_obj.getUniqueChildren()
1058 return queryset.filter(location__in=children)
1060 tags = common.filters.TagsFilter(label=_('Tags'))
1063class StockApiMixin(SerializerContextMixin):
1064 """Mixin class for StockItem API endpoints."""
1066 serializer_class = StockSerializers.StockItemSerializer
1067 queryset = StockItem.objects.all()
1069 def get_queryset(self, *args, **kwargs):
1070 """Annotate queryset."""
1071 queryset = super().get_queryset(*args, **kwargs)
1072 queryset = StockSerializers.StockItemSerializer.annotate_queryset(queryset)
1074 return queryset
1076 def get_serializer_context(self):
1077 """Extend serializer context."""
1078 ctx = super().get_serializer_context()
1079 ctx['user'] = getattr(self.request, 'user', None)
1081 return ctx
1084class StockOutputOptions(OutputConfiguration):
1085 """Output options for StockItem serializers."""
1087 OPTIONS = [
1088 InvenTreeOutputOption('part_detail', default=True),
1089 InvenTreeOutputOption('path_detail'),
1090 InvenTreeOutputOption('supplier_part_detail'),
1091 InvenTreeOutputOption('location_detail'),
1092 InvenTreeOutputOption('tests'),
1093 ]
1096class StockList(
1097 DataExportViewMixin,
1098 BulkUpdateMixin,
1099 StockApiMixin,
1100 OutputOptionsMixin,
1101 ListCreateDestroyAPIView,
1102):
1103 """API endpoint for list view of Stock objects.
1105 - GET: Return a list of all StockItem objects (with optional query filters)
1106 - POST: Create a new StockItem
1107 - DELETE: Delete multiple StockItem objects
1108 """
1110 filterset_class = StockFilter
1111 output_options = StockOutputOptions
1113 def create(self, request, *args, **kwargs):
1114 """Create a new StockItem object via the API.
1116 We override the default 'create' implementation.
1118 If a location is *not* specified, but the linked *part* has a default location,
1119 we can pre-fill the location automatically.
1120 """
1121 user = request.user
1123 # Copy the request data, to side-step "mutability" issues
1124 data = OrderedDict()
1125 # Update with cleaned input data
1126 data.update(self.clean_data(request.data))
1128 quantity = data.get('quantity', None)
1130 if quantity is None:
1131 raise ValidationError({'quantity': _('Quantity is required')})
1133 try:
1134 part = Part.objects.get(pk=data.get('part', None))
1135 except (ValueError, Part.DoesNotExist):
1136 raise ValidationError({'part': _('Valid part must be supplied')})
1138 location = data.get('location', None)
1139 # Override location if not specified
1140 if location is None and part.default_location:
1141 data['location'] = part.default_location.pk
1143 expiry_date = data.get('expiry_date', None)
1145 # An expiry date was *not* specified - try to infer it!
1146 if expiry_date is None and part.default_expiry > 0:
1147 data['expiry_date'] = InvenTree.helpers.current_date() + timedelta(
1148 days=part.default_expiry
1149 )
1151 # Attempt to extract serial numbers from submitted data
1152 serials = None
1154 # Check if a set of serial numbers was provided
1155 serial_numbers = data.pop('serial_numbers', '')
1157 # Exclude 'serial' from submitted data
1158 # We use 'serial_numbers' for item creation
1159 data.pop('serial', None)
1161 # Check if the supplier_part has a package size defined, which is not 1
1162 if supplier_part_id := data.get('supplier_part', None):
1163 try:
1164 supplier_part = SupplierPart.objects.get(pk=supplier_part_id)
1165 except Exception:
1166 raise ValidationError({
1167 'supplier_part': _('The given supplier part does not exist')
1168 })
1170 if supplier_part.base_quantity() != 1:
1171 # Skip this check if pack size is 1 - makes no difference
1172 # use_pack_size = True -> Multiply quantity by pack size
1173 # use_pack_size = False -> Use quantity as is
1174 if 'use_pack_size' not in data:
1175 raise ValidationError({
1176 'use_pack_size': _(
1177 'The supplier part has a pack size defined, but flag use_pack_size not set'
1178 )
1179 })
1180 elif bool(data.get('use_pack_size')):
1181 quantity = data['quantity'] = supplier_part.base_quantity(quantity)
1183 # Divide purchase price by pack size, to save correct price per stock item
1184 if (
1185 data.get('purchase_price')
1186 and supplier_part.pack_quantity_native
1187 ):
1188 try:
1189 data['purchase_price'] = float(
1190 data['purchase_price']
1191 ) / float(supplier_part.pack_quantity_native)
1192 except ValueError: # pragma: no cover
1193 pass
1195 # Now remove the flag from data, so that it doesn't interfere with saving
1196 # Do this regardless of results above
1197 data.pop('use_pack_size', None)
1199 # Extract 'status' flag from data
1200 status_raw = data.pop('status', None)
1201 status_custom = data.pop('status_custom_key', None)
1202 status_value = status_custom or status_raw
1204 # Assign serial numbers for a trackable part
1205 if serial_numbers:
1206 if not part.trackable:
1207 raise ValidationError({
1208 'serial_numbers': [
1209 _('Serial numbers cannot be supplied for a non-trackable part')
1210 ]
1211 })
1213 # If serial numbers are specified, check that they match!
1214 try:
1215 serials = extract_serial_numbers(
1216 serial_numbers, quantity, part.get_latest_serial_number(), part=part
1217 )
1219 # Determine if any of the specified serial numbers are invalid
1220 # Note "invalid" means either they already exist, or do not pass custom rules
1221 invalid = []
1222 errors = []
1224 try:
1225 invalid = part.find_conflicting_serial_numbers(serials)
1226 except DjangoValidationError as exc:
1227 errors.append(exc.message)
1229 if len(invalid) > 0:
1230 msg = _('The following serial numbers already exist or are invalid')
1231 msg += ' : '
1232 msg += ','.join([str(e) for e in invalid])
1234 errors.append(msg)
1236 if len(errors) > 0:
1237 raise ValidationError({'serial_numbers': errors})
1239 except DjangoValidationError as e:
1240 raise ValidationError({
1241 'quantity': e.messages,
1242 'serial_numbers': e.messages,
1243 })
1245 if serials is not None:
1246 """If the stock item is going to be serialized, set the quantity to 1."""
1247 data['quantity'] = 1
1249 # De-serialize the provided data
1250 serializer = self.get_serializer(data=data)
1251 serializer.is_valid(raise_exception=True)
1253 # Extract location information
1254 location = serializer.validated_data.get('location', None)
1256 with transaction.atomic():
1257 if serials:
1258 # Create multiple serialized StockItem objects
1259 items = StockItem._create_serial_numbers(
1260 serials, **serializer.validated_data
1261 )
1263 # Next, bulk-create stock tracking entries for the newly created items
1264 tracking = []
1266 for item in items:
1267 if status_value and not item.compare_status(status_value):
1268 item.set_status(status_value)
1269 item.save()
1271 if entry := item.add_tracking_entry(
1272 StockHistoryCode.CREATED,
1273 user,
1274 deltas={'status': status_value},
1275 location=location,
1276 quantity=float(item.quantity),
1277 commit=False,
1278 ):
1279 tracking.append(entry)
1281 StockItemTracking.objects.bulk_create(tracking, batch_size=250)
1283 # Annotate the stock items with part information
1284 queryset = StockSerializers.StockItemSerializer.annotate_queryset(items)
1286 response = StockSerializers.StockItemSerializer(
1287 queryset, many=True, context=self.get_serializer_context()
1288 )
1290 response_data = response.data
1292 else:
1293 # Create a single StockItem object
1294 # Note: This automatically creates a tracking entry
1295 item = StockItem(**serializer.validated_data)
1297 if status_value and not item.compare_status(status_value):
1298 item.set_status(status_value)
1300 item.save(user=user)
1301 item.refresh_from_db()
1303 response_data = [
1304 StockSerializers.StockItemSerializer(
1305 item, context=self.get_serializer_context()
1306 ).data
1307 ]
1309 return Response(
1310 response_data,
1311 status=status.HTTP_201_CREATED,
1312 headers=self.get_success_headers(serializer.data),
1313 )
1315 filter_backends = SEARCH_ORDER_FILTER
1317 ordering_field_aliases = {
1318 'part': 'part__name',
1319 'location': 'location__pathstring',
1320 'IPN': 'part__IPN',
1321 'SKU': 'supplier_part__SKU',
1322 'MPN': 'supplier_part__manufacturer_part__MPN',
1323 'stock': ['quantity', 'serial_int', 'serial'],
1324 }
1326 ordering_fields = [
1327 'batch',
1328 'location',
1329 'part',
1330 'part__name',
1331 'part__IPN',
1332 'updated',
1333 'purchase_price',
1334 'creation_date',
1335 'stocktake_date',
1336 'expiry_date',
1337 'packaging',
1338 'quantity',
1339 'stock',
1340 'status',
1341 'IPN',
1342 'SKU',
1343 'MPN',
1344 ]
1346 ordering = ['part__name', 'quantity', 'location']
1348 search_fields = [
1349 'serial',
1350 'batch',
1351 'location__name',
1352 'part__name',
1353 'part__IPN',
1354 'part__description',
1355 'supplier_part__SKU',
1356 'supplier_part__supplier__name',
1357 'supplier_part__manufacturer_part__MPN',
1358 'supplier_part__manufacturer_part__manufacturer__name',
1359 'tags__name',
1360 'tags__slug',
1361 ]
1364class StockDetail(StockApiMixin, OutputOptionsMixin, RetrieveUpdateDestroyAPI):
1365 """API detail endpoint for a single StockItem instance."""
1367 output_options = StockOutputOptions
1370class StockItemSerialNumbers(RetrieveAPI):
1371 """View extra serial number information for a given stock item.
1373 Provides information on the "previous" and "next" stock items,
1374 based on the serial number of the given stock item.
1375 """
1377 queryset = StockItem.objects.all()
1378 serializer_class = StockSerializers.StockItemSerialNumbersSerializer
1381class StockItemTestResultMixin(SerializerContextMixin):
1382 """Mixin class for the StockItemTestResult API endpoints."""
1384 queryset = StockItemTestResult.objects.all()
1385 serializer_class = StockSerializers.StockItemTestResultSerializer
1387 def get_serializer_context(self):
1388 """Extend serializer context."""
1389 ctx = super().get_serializer_context()
1390 ctx['request'] = self.request
1391 return ctx
1394class StockItemTestResultOutputOptions(OutputConfiguration):
1395 """Output options for StockItemTestResult endpoint."""
1397 OPTIONS = [
1398 InvenTreeOutputOption(flag='user_detail'),
1399 InvenTreeOutputOption(flag='template_detail'),
1400 ]
1403class StockItemTestResultDetail(
1404 StockItemTestResultMixin, OutputOptionsMixin, RetrieveUpdateDestroyAPI
1405):
1406 """Detail endpoint for StockItemTestResult."""
1408 output_options = StockItemTestResultOutputOptions
1411class StockItemTestResultFilter(FilterSet):
1412 """API filter for the StockItemTestResult list."""
1414 class Meta:
1415 """Metaclass options."""
1417 model = StockItemTestResult
1419 # Simple filter fields
1420 fields = ['user', 'template', 'result', 'value']
1422 build = rest_filters.ModelChoiceFilter(
1423 label='Build', queryset=Build.objects.all(), field_name='stock_item__build'
1424 )
1426 part = rest_filters.ModelChoiceFilter(
1427 label='Part', queryset=Part.objects.all(), field_name='stock_item__part'
1428 )
1430 required = rest_filters.BooleanFilter(
1431 label='Required', field_name='template__required'
1432 )
1434 enabled = rest_filters.BooleanFilter(
1435 label='Enabled', field_name='template__enabled'
1436 )
1438 test = rest_filters.CharFilter(
1439 label='Test name (case insensitive)', method='filter_test_name'
1440 )
1442 def filter_test_name(self, queryset, name, value):
1443 """Filter by test name.
1445 This method is provided for legacy support,
1446 where the StockItemTestResult model had a "test" field.
1447 Now the "test" name is stored against the PartTestTemplate model
1448 """
1449 key = generateTestKey(value)
1450 return queryset.filter(template__key=key)
1452 include_installed = rest_filters.BooleanFilter(
1453 method='filter_include_installed',
1454 label=_('Include Installed'),
1455 help_text=_(
1456 'If true, include test results for items installed underneath the given stock item'
1457 ),
1458 )
1460 stock_item = rest_filters.NumberFilter(
1461 method='filter_stock_item',
1462 label=_('Stock Item'),
1463 help_text=_('Filter by numeric Stock Item ID'),
1464 )
1466 def filter_include_installed(self, queryset, name, value):
1467 """Dummy filter method for 'include_installed'.
1469 - Ensures 'include_installed' appears in API documentation
1470 - Does NOT actually filter the queryset directly
1471 - The actual logic is handled in filter_stock_item method
1472 """
1473 return queryset
1475 def filter_stock_item(self, queryset, name, value):
1476 """Filter for stock_item that also applies include_installed logic."""
1477 include_installed = str2bool(self.data.get('include_installed', False))
1479 try:
1480 item = StockItem.objects.get(pk=value)
1482 except StockItem.DoesNotExist:
1483 raise ValidationError({
1484 'stock_item': _('Stock item with ID {id} does not exist').format(
1485 id=value
1486 )
1487 })
1489 items = [item]
1491 if include_installed:
1492 # Include items which are installed "underneath" this item
1493 # Note that this function is recursive!
1494 installed_items = item.get_installed_items(cascade=True)
1495 items += list(installed_items)
1497 return queryset.filter(stock_item__in=items)
1500class StockItemTestResultList(
1501 BulkCreateMixin,
1502 StockItemTestResultMixin,
1503 OutputOptionsMixin,
1504 ListCreateDestroyAPIView,
1505):
1506 """API endpoint for listing (and creating) a StockItemTestResult object."""
1508 filterset_class = StockItemTestResultFilter
1509 filter_backends = SEARCH_ORDER_FILTER
1510 output_options = StockItemTestResultOutputOptions
1512 filterset_fields = ['user', 'template', 'result', 'value']
1513 ordering_fields = [
1514 'date',
1515 'result',
1516 'started_datetime',
1517 'finished_datetime',
1518 'test_station',
1519 ]
1521 ordering = 'date'
1523 def perform_create(self, serializer):
1524 """Create a new test result object.
1526 Also, check if an attachment was uploaded alongside the test result,
1527 and save it to the database if it were.
1528 """
1529 serializer.save(user=self.request.user)
1532class StockTrackingDetail(RetrieveAPI):
1533 """Detail API endpoint for StockItemTracking model."""
1535 queryset = StockItemTracking.objects.all()
1536 serializer_class = StockSerializers.StockTrackingSerializer
1539class StockTrackingOutputOptions(OutputConfiguration):
1540 """Output options for StockItemTracking endpoint."""
1542 OPTIONS = [
1543 InvenTreeOutputOption(flag='item_detail'),
1544 InvenTreeOutputOption(flag='user_detail'),
1545 ]
1548class StockTrackingFilter(FilterSet):
1549 """API filter options for the StockTrackingList endpoint."""
1551 class Meta:
1552 """Metaclass options."""
1554 model = StockItemTracking
1555 fields = ['item', 'user']
1557 include_variants = rest_filters.BooleanFilter(
1558 label=_('Include Part Variants'), method='filter_include_variants'
1559 )
1561 def filter_include_variants(self, queryset, name, value):
1562 """Filter by whether or not to include part variants.
1564 Note:
1565 - This filter does nothing by itself, and is only used to modify the behavior of the 'part' filter.
1566 - Refer to the 'filter_part' method for more information on how this works.
1567 """
1568 return queryset
1570 part = rest_filters.ModelChoiceFilter(
1571 label=_('Part'), queryset=Part.objects.all(), method='filter_part'
1572 )
1574 def filter_part(self, queryset, name, part):
1575 """Filter StockTracking entries by the linked part.
1577 Note:
1578 - This filter behavior also takes into account the 'include_variants' filter, which determines whether or not to include part variants in the results.
1579 """
1580 include_variants = str2bool(self.data.get('include_variants', False))
1582 if include_variants:
1583 return queryset.filter(part__in=part.get_descendants(include_self=True))
1584 else:
1585 return queryset.filter(part=part)
1587 min_date = InvenTreeDateFilter(
1588 label=_('Date after'), field_name='date', lookup_expr='gt'
1589 )
1591 max_date = InvenTreeDateFilter(
1592 label=_('Date before'), field_name='date', lookup_expr='lt'
1593 )
1596class StockTrackingList(
1597 SerializerContextMixin, DataExportViewMixin, OutputOptionsMixin, ListAPI
1598):
1599 """API endpoint for list view of StockItemTracking objects.
1601 StockItemTracking objects are read-only
1602 (they are created by internal model functionality)
1604 - GET: Return list of StockItemTracking objects
1605 """
1607 queryset = StockItemTracking.objects.all().prefetch_related('item', 'part')
1608 serializer_class = StockSerializers.StockTrackingSerializer
1609 filterset_class = StockTrackingFilter
1610 output_options = StockTrackingOutputOptions
1612 def get_delta_model_map(self) -> dict:
1613 """Return a mapping of delta models to their respective models and serializers.
1615 This is used to generate additional context information for the historical data,
1616 with some attempt at caching so that we can reduce the number of database hits.
1617 """
1618 return {
1619 'part': (Part, PartBriefSerializer),
1620 'location': (StockLocation, StockSerializers.LocationSerializer),
1621 'customer': (Company, CompanySerializer),
1622 'purchaseorder': (PurchaseOrder, PurchaseOrderSerializer),
1623 'salesorder': (SalesOrder, SalesOrderSerializer),
1624 'returnorder': (ReturnOrder, ReturnOrderSerializer),
1625 'transferorder': (TransferOrder, TransferOrderSerializer),
1626 'buildorder': (Build, BuildSerializer),
1627 'item': (StockItem, StockSerializers.StockItemSerializer),
1628 'stockitem': (StockItem, StockSerializers.StockItemSerializer),
1629 }
1631 def list(self, request, *args, **kwargs):
1632 """List all stock tracking entries."""
1633 queryset = self.filter_queryset(self.get_queryset())
1635 page = self.paginate_queryset(queryset)
1637 if page is not None:
1638 serializer = self.get_serializer(page, many=True)
1639 else:
1640 serializer = self.get_serializer(queryset, many=True)
1642 data = serializer.data
1644 delta_models = self.get_delta_model_map()
1646 # Construct a set of related models we need to lookup for later
1647 related_model_lookups = {key: set() for key in delta_models}
1649 # Run a first pass through the data to determine which related models we need to lookup
1650 for item in data: 1650 ↛ 1651line 1650 didn't jump to line 1651 because the loop on line 1650 never started
1651 deltas = item['deltas'] or {}
1653 for key in delta_models:
1654 if key in deltas:
1655 related_model_lookups[key].add(deltas[key])
1657 for key in delta_models:
1658 model, serializer = delta_models[key]
1660 # Fetch all related models in one go
1661 related_models = model.objects.filter(pk__in=related_model_lookups[key])
1663 # Construct a mapping of pk -> serialized data
1664 related_data = {obj.pk: serializer(obj).data for obj in related_models}
1666 # Now, update the data with the serialized data
1667 for item in data: 1667 ↛ 1668line 1667 didn't jump to line 1668 because the loop on line 1667 never started
1668 deltas = item['deltas'] or {}
1670 if key in deltas:
1671 item['deltas'][f'{key}_detail'] = related_data.get(deltas[key])
1673 if page is not None:
1674 return self.get_paginated_response(data)
1676 return Response(data)
1678 filter_backends = SEARCH_ORDER_FILTER
1680 ordering = '-date'
1682 ordering_fields = ['date']
1684 search_fields = ['notes']
1687stock_api_urls = [
1688 path(
1689 'location/',
1690 include([
1691 path('tree/', StockLocationTree.as_view(), name='api-location-tree'),
1692 # Stock location detail endpoints
1693 path(
1694 '<int:pk>/',
1695 include([
1696 meta_path(StockLocation),
1697 path('', StockLocationDetail.as_view(), name='api-location-detail'),
1698 ]),
1699 ),
1700 path('', StockLocationList.as_view(), name='api-location-list'),
1701 ]),
1702 ),
1703 # Stock location type endpoints
1704 path(
1705 'location-type/',
1706 include([
1707 path(
1708 '<int:pk>/',
1709 include([
1710 meta_path(StockLocationType),
1711 path(
1712 '',
1713 StockLocationTypeDetail.as_view(),
1714 name='api-location-type-detail',
1715 ),
1716 ]),
1717 ),
1718 path('', StockLocationTypeList.as_view(), name='api-location-type-list'),
1719 ]),
1720 ),
1721 # Endpoints for bulk stock adjustment actions
1722 path('count/', StockCount.as_view(), name='api-stock-count'),
1723 path('add/', StockAdd.as_view(), name='api-stock-add'),
1724 path('remove/', StockRemove.as_view(), name='api-stock-remove'),
1725 path('transfer/', StockTransfer.as_view(), name='api-stock-transfer'),
1726 path('return/', StockReturn.as_view(), name='api-stock-return'),
1727 path('assign/', StockAssign.as_view(), name='api-stock-assign'),
1728 path('merge/', StockMerge.as_view(), name='api-stock-merge'),
1729 path('change_status/', StockChangeStatus.as_view(), name='api-stock-change-status'),
1730 # StockItemTestResult API endpoints
1731 path(
1732 'test/',
1733 include([
1734 path(
1735 '<int:pk>/',
1736 include([
1737 meta_path(StockItemTestResult),
1738 path(
1739 '',
1740 StockItemTestResultDetail.as_view(),
1741 name='api-stock-test-result-detail',
1742 ),
1743 ]),
1744 ),
1745 path(
1746 '', StockItemTestResultList.as_view(), name='api-stock-test-result-list'
1747 ),
1748 ]),
1749 ),
1750 # StockItemTracking API endpoints
1751 path(
1752 'track/',
1753 include([
1754 path(
1755 '<int:pk>/',
1756 StockTrackingDetail.as_view(),
1757 name='api-stock-tracking-detail',
1758 ),
1759 # Stock tracking status code information
1760 path(
1761 'status/',
1762 StatusView.as_view(),
1763 {StatusView.MODEL_REF: StockHistoryCode},
1764 name='api-stock-tracking-status-codes',
1765 ),
1766 path('', StockTrackingList.as_view(), name='api-stock-tracking-list'),
1767 ]),
1768 ),
1769 # Detail views for a single stock item
1770 path(
1771 '<int:pk>/',
1772 include([
1773 path('convert/', StockItemConvert.as_view(), name='api-stock-item-convert'),
1774 path('install/', StockItemInstall.as_view(), name='api-stock-item-install'),
1775 meta_path(StockItem),
1776 path(
1777 'serialize/',
1778 StockItemSerialize.as_view(),
1779 name='api-stock-item-serialize',
1780 ),
1781 path(
1782 'uninstall/',
1783 StockItemUninstall.as_view(),
1784 name='api-stock-item-uninstall',
1785 ),
1786 path(
1787 'serial-numbers/',
1788 StockItemSerialNumbers.as_view(),
1789 name='api-stock-item-serial-numbers',
1790 ),
1791 path('', StockDetail.as_view(), name='api-stock-detail'),
1792 ]),
1793 ),
1794 # Stock item status code information
1795 path(
1796 'status/',
1797 StatusView.as_view(),
1798 {StatusView.MODEL_REF: StockStatus},
1799 name='api-stock-status-codes',
1800 ),
1801 # Anything else
1802 path('', StockList.as_view(), name='api-stock-list'),
1803]