Coverage for src/backend/InvenTree/stock/models.py: 19%
1235 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 17:47 +0000
1"""Stock database model definitions."""
3from __future__ import annotations
5import os
6from datetime import timedelta
7from decimal import Decimal, InvalidOperation
9from django.conf import settings
10from django.contrib.auth.models import User
11from django.core.exceptions import ValidationError
12from django.core.validators import MinValueValidator
13from django.db import models, transaction
14from django.db.models import Q, QuerySet, Sum
15from django.db.models.functions import Coalesce
16from django.db.models.signals import post_delete, post_save
17from django.db.utils import IntegrityError, OperationalError
18from django.dispatch import receiver
19from django.urls import reverse
20from django.utils.translation import gettext_lazy as _
22import structlog
23from djmoney.contrib.exchange.models import convert_money
24from mptt.managers import TreeManager
25from mptt.models import TreeForeignKey
27import build.models
28import common.models
29import InvenTree.exceptions
30import InvenTree.helpers
31import InvenTree.models
32import InvenTree.ready
33import InvenTree.tasks
34import order.models
35import report.mixins
36import stock.tasks
37from common.icons import validate_icon
38from common.settings import get_global_setting
39from company import models as CompanyModels
40from generic.enums import StringEnum
41from generic.states import StatusCodeMixin
42from generic.states.fields import InvenTreeCustomStatusModelField
43from InvenTree.fields import InvenTreeModelMoneyField, InvenTreeURLField
44from InvenTree.status_codes import (
45 SalesOrderStatusGroups,
46 StockHistoryCode,
47 StockStatus,
48 StockStatusGroups,
49)
50from order.status_codes import TransferOrderStatusGroups
51from part import models as PartModels
52from plugin.events import trigger_event
53from stock.events import StockEvents
54from stock.generators import generate_batch_code
55from users.models import Owner
57logger = structlog.get_logger('inventree')
60class StockLocationType(InvenTree.models.MetadataMixin, models.Model):
61 """A type of stock location like Warehouse, room, shelf, drawer.
63 Attributes:
64 name: brief name
65 description: longer form description
66 icon: icon class
67 """
69 IMPORT_ID_FIELDS = ['name']
71 class Meta:
72 """Metaclass defines extra model properties."""
74 verbose_name = _('Stock Location type')
75 verbose_name_plural = _('Stock Location types')
77 @staticmethod
78 def get_api_url():
79 """Return API url."""
80 return reverse('api-location-type-list')
82 def __str__(self):
83 """String representation of a StockLocationType."""
84 return self.name
86 name = models.CharField(
87 blank=False, max_length=100, verbose_name=_('Name'), help_text=_('Name')
88 )
90 description = models.CharField(
91 blank=True,
92 max_length=250,
93 verbose_name=_('Description'),
94 help_text=_('Description (optional)'),
95 )
97 icon = models.CharField(
98 blank=True,
99 max_length=100,
100 verbose_name=_('Icon'),
101 help_text=_('Default icon for all locations that have no icon set (optional)'),
102 validators=[validate_icon],
103 )
106class StockLocationReportContext(report.mixins.BaseReportContext):
107 """Report context for the StockLocation model.
109 Attributes:
110 location: The StockLocation object itself
111 qr_data: Formatted QR code data for the StockLocation
112 parent: The parent StockLocation object
113 stock_location: The StockLocation object itself (shadow of 'location')
114 stock_items: Query set of all StockItem objects which are located in the StockLocation
115 """
117 location: StockLocation
118 qr_data: str
119 parent: StockLocation | None
120 stock_location: StockLocation
121 stock_items: report.mixins.QuerySet[StockItem]
124class StockLocation(
125 InvenTree.models.PluginValidationMixin,
126 InvenTree.models.InvenTreeParameterMixin,
127 InvenTree.models.InvenTreeBarcodeMixin,
128 InvenTree.models.InvenTreeTagsMixin,
129 report.mixins.InvenTreeReportMixin,
130 InvenTree.models.PathStringMixin,
131 InvenTree.models.MetadataMixin,
132 InvenTree.models.InvenTreeTree,
133):
134 """Organization tree for StockItem objects.
136 A "StockLocation" can be considered a warehouse, or storage location
137 Stock locations can be hierarchical as required
138 """
140 ITEM_PARENT_KEY = 'location'
141 EXTRA_PATH_FIELDS = ['icon']
142 IMPORT_ID_FIELDS = ['pathstring', 'name']
144 objects = TreeManager()
146 class Meta:
147 """Metaclass defines extra model properties."""
149 verbose_name = _('Stock Location')
150 verbose_name_plural = _('Stock Locations')
152 def delete(self, *args, **kwargs):
153 """Custom model deletion routine, which updates any child locations or items.
155 This must be handled within a transaction.atomic(), otherwise the tree structure is damaged
156 """
157 super().delete(
158 delete_children=kwargs.get('delete_sub_locations', False),
159 delete_items=kwargs.get('delete_stock_items', False),
160 )
162 @staticmethod
163 def get_api_url():
164 """Return API url."""
165 return reverse('api-location-list')
167 @classmethod
168 def barcode_model_type_code(cls):
169 """Return the associated barcode model type code for this model."""
170 return 'SL'
172 def report_context(self) -> StockLocationReportContext:
173 """Return report context data for this StockLocation."""
174 return {
175 'location': self,
176 'qr_data': self.barcode,
177 'parent': self.parent,
178 'stock_location': self,
179 'stock_items': self.get_stock_items(),
180 }
182 custom_icon = models.CharField(
183 blank=True,
184 null=True,
185 max_length=100,
186 verbose_name=_('Icon'),
187 help_text=_('Icon (optional)'),
188 db_column='icon',
189 validators=[validate_icon],
190 )
192 owner = models.ForeignKey(
193 Owner,
194 on_delete=models.SET_NULL,
195 blank=True,
196 null=True,
197 verbose_name=_('Owner'),
198 help_text=_('Select Owner'),
199 related_name='stock_locations',
200 )
202 structural = models.BooleanField(
203 default=False,
204 verbose_name=_('Structural'),
205 help_text=_(
206 'Stock items may not be directly located into a structural stock locations, '
207 'but may be located to child locations.'
208 ),
209 )
211 external = models.BooleanField(
212 default=False,
213 verbose_name=_('External'),
214 help_text=_('This is an external stock location'),
215 )
217 location_type = models.ForeignKey(
218 StockLocationType,
219 on_delete=models.SET_NULL,
220 verbose_name=_('Location type'),
221 related_name='stock_locations',
222 null=True,
223 blank=True,
224 help_text=_('Stock location type of this location'),
225 )
227 @property
228 def icon(self) -> str:
229 """Get the current icon used for this location.
231 The icon field on this model takes precedences over the possibly assigned stock location type
232 """
233 if self.custom_icon: 233 ↛ 234line 233 didn't jump to line 234 because the condition on line 233 was never true
234 return self.custom_icon
236 if self.location_type: 236 ↛ 237line 236 didn't jump to line 237 because the condition on line 236 was never true
237 return self.location_type.icon
239 if default_icon := get_global_setting( 239 ↛ 242line 239 didn't jump to line 242 because the condition on line 239 was never true
240 'STOCK_LOCATION_DEFAULT_ICON', cache=True
241 ):
242 return default_icon
244 return ''
246 @icon.setter
247 def icon(self, value):
248 """Setter to keep model API compatibility.
250 But be careful:
251 If the field gets loaded as default value by any form which is later saved,
252 the location no longer inherits its icon from the location type.
253 """
254 self.custom_icon = value
256 def get_location_owner(self):
257 """Get the closest "owner" for this location.
259 Start at this location, and traverse "up" the location tree until we find an owner
260 """
261 for loc in self.get_ancestors(include_self=True, ascending=True):
262 if loc.owner is not None:
263 return loc.owner
265 return None
267 def check_ownership(self, user):
268 """Check if the user "owns" (is one of the owners of) the location."""
269 # Superuser accounts automatically "own" everything
270 if user.is_superuser:
271 return True
273 ownership_enabled = get_global_setting('STOCK_OWNERSHIP_CONTROL')
275 if not ownership_enabled:
276 # Location ownership function is not enabled, so return True
277 return True
279 owner = self.get_location_owner()
281 if owner is None:
282 # No owner set, for this location or any location above
283 # So, no ownership checks to perform!
284 return True
286 return owner.is_user_allowed(user, include_group=True)
288 def clean(self):
289 """Custom clean action for the StockLocation model.
291 Ensure stock location can't be made structural if stock items already located to them
292 """
293 if self.pk and self.structural and self.stock_item_count(False) > 0: 293 ↛ 294line 293 didn't jump to line 294 because the condition on line 293 was never true
294 raise ValidationError(
295 _(
296 'You cannot make this stock location structural because some stock items '
297 'are already located into it!'
298 )
299 )
300 super().clean()
302 def get_absolute_url(self):
303 """Return url for instance."""
304 return InvenTree.helpers.pui_url(f'/stock/location/{self.id}')
306 def get_stock_items(self, cascade=True):
307 """Return a queryset for all stock items under this category.
309 Args:
310 cascade: If True, also look under sublocations (default = True)
311 """
312 if cascade:
313 query = StockItem.objects.filter(
314 location__in=self.getUniqueChildren(include_self=True)
315 )
316 else:
317 query = StockItem.objects.filter(location=self.pk)
319 return query
321 def stock_item_count(self, cascade=True):
322 """Return the number of StockItem objects which live in or under this category."""
323 return self.get_stock_items(cascade).count()
325 @property
326 def item_count(self):
327 """Simply returns the number of stock items in this location.
329 Required for tree view serializer.
330 """
331 return self.stock_item_count()
333 def get_items(self, cascade=False):
334 """Return a queryset for all stock items under this category."""
335 return self.get_stock_items(cascade=cascade)
338def default_delete_on_deplete():
339 """Return a default value for the 'delete_on_deplete' field.
341 Prior to 2022-12-24, this field was set to True by default.
342 Now, there is a user-configurable setting to govern default behavior.
343 """
344 try:
345 return get_global_setting('STOCK_DELETE_DEPLETED_DEFAULT', True)
346 except (IntegrityError, OperationalError):
347 # Revert to original default behavior
348 return True
351class StockItemReportContext(report.mixins.BaseReportContext):
352 """Report context for the StockItem model.
354 Attributes:
355 barcode_data: Generated barcode data for the StockItem
356 barcode_hash: Hash of the barcode data
357 batch: The batch code for the StockItem
358 child_items: Query set of all StockItem objects which are children of this StockItem
359 ipn: The IPN (internal part number) of the associated Part
360 installed_items: Query set of all StockItem objects which are installed in this StockItem
361 item: The StockItem object itself
362 name: The name of the associated Part
363 part: The Part object which is associated with the StockItem
364 qr_data: Generated QR code data for the StockItem
365 qr_url: Generated URL for embedding in a QR code
366 parameters: Dict object containing the parameters associated with the base Part
367 quantity: The quantity of the StockItem
368 result_list: FLattened list of TestResult data associated with the stock item
369 results: Dict object of TestResult data associated with the StockItem
370 serial: The serial number of the StockItem
371 stock_item: The StockItem object itself (shadow of 'item')
372 tests: Dict object of TestResult data associated with the StockItem (shadow of 'results')
373 test_keys: List of test keys associated with the StockItem
374 test_template_list: List of test templates associated with the StockItem
375 test_templates: Dict object of test templates associated with the StockItem
376 """
378 barcode_data: str
379 barcode_hash: str
380 batch: str
381 child_items: report.mixins.QuerySet[StockItem]
382 ipn: str | None
383 installed_items: set[StockItem]
384 item: StockItem
385 name: str
386 part: PartModels.Part
387 qr_data: str
388 qr_url: str
389 parameters: dict[str, str]
390 quantity: Decimal
391 result_list: list[StockItemTestResult]
392 results: dict[str, StockItemTestResult]
393 serial: str | None
394 stock_item: StockItem
395 tests: dict[str, StockItemTestResult]
396 test_keys: list[str]
397 test_template_list: report.mixins.QuerySet[PartModels.PartTestTemplate]
398 test_templates: dict[str, PartModels.PartTestTemplate]
401class StockSortOrder(StringEnum):
402 """Enum of ORM sort fields available for stock auto-allocation."""
404 DATE_OLDEST = 'updated'
405 DATE_NEWEST = '-updated'
406 QUANTITY_ASC = 'quantity'
407 QUANTITY_DESC = '-quantity'
408 EXPIRY_SOONEST = 'expiry_date'
411STOCK_SORT_CHOICES = [
412 (StockSortOrder.DATE_OLDEST, _('Oldest stock first (FIFO)')),
413 (StockSortOrder.DATE_NEWEST, _('Newest stock first (LIFO)')),
414 (StockSortOrder.QUANTITY_ASC, _('Smallest quantity first')),
415 (StockSortOrder.QUANTITY_DESC, _('Largest quantity first')),
416 (StockSortOrder.EXPIRY_SOONEST, _('Soonest expiry date first')),
417]
419STOCK_SORT_DEFAULT = StockSortOrder.DATE_OLDEST
422class StockItem(
423 InvenTree.models.PluginValidationMixin,
424 InvenTree.models.InvenTreeAttachmentMixin,
425 InvenTree.models.InvenTreeBarcodeMixin,
426 InvenTree.models.InvenTreeNotesMixin,
427 InvenTree.models.InvenTreeTagsMixin,
428 StatusCodeMixin,
429 report.mixins.InvenTreeReportMixin,
430 common.models.MetaMixin,
431 InvenTree.models.MetadataMixin,
432 InvenTree.models.InvenTreeTree,
433):
434 """A StockItem object represents a quantity of physical instances of a part.
436 Attributes:
437 parent: Link to another StockItem from which this StockItem was created
438 part: Link to the master abstract part that this StockItem is an instance of
439 supplier_part: Link to a specific SupplierPart (optional)
440 location: Where this StockItem is located
441 quantity: Number of stocked units
442 batch: Batch number for this StockItem
443 serial: Unique serial number for this StockItem
444 link: Optional URL to link to external resource
445 creation_date: Date that this stock item was created (auto)
446 updated: Date that the quantity of this stock item was last updated (auto)
447 expiry_date: Expiry date of the StockItem (optional)
448 stocktake_date: Date of last stocktake for this item
449 stocktake_user: User that performed the most recent stocktake
450 delete_on_deplete: If True, StockItem will be deleted when the stock level gets to zero
451 status: Status of this StockItem (ref: stock.status_codes.StockStatus)
452 notes: Extra notes field
453 build: Link to a Build (if this stock item was created from a build)
454 is_building: Boolean field indicating if this stock item is currently being built (or is "in production")
455 purchase_order: Link to a PurchaseOrder (if this stock item was created from a PurchaseOrder)
456 sales_order: Link to a SalesOrder object (if the StockItem has been assigned to a SalesOrder)
457 purchase_price: The unit purchase price for this StockItem - this is the unit price at time of purchase (if this item was purchased from an external supplier)
458 packaging: Description of how the StockItem is packaged (e.g. "reel", "loose", "tape" etc)
459 """
461 IMPORT_ID_FIELDS = ['serial']
462 STATUS_CLASS = StockStatus
464 class Meta:
465 """Model meta options."""
467 verbose_name = _('Stock Item')
469 class MPTTMeta:
470 """MPTT metaclass options."""
472 order_insertion_by = ['part']
474 def save(self, *args, **kwargs):
475 """Save this StockItem to the database.
477 Performs a number of checks:
478 - Unique serial number requirement
479 - Adds a transaction note when the item is first created.
480 """
481 self.validate_unique()
482 self.clean()
483 self.update_serial_number()
485 user = kwargs.pop('user', None)
487 if user is None:
488 user = getattr(self, '_user', None)
490 # If 'add_note = False' specified, then no tracking note will be added for item creation
491 add_note = kwargs.pop('add_note', True)
493 notes = kwargs.pop('notes', '')
495 if self.pk:
496 # StockItem has already been saved
498 # Check if "interesting" fields have been changed
499 # (we wish to record these as historical records)
501 try:
502 old = StockItem.objects.get(pk=self.pk)
503 old_custom_status = old.get_custom_status()
504 custom_status = self.get_custom_status()
506 deltas = {}
508 # Status changed?
509 if old.status != self.status:
510 # Custom status changed?
511 # Matches custom status tracking behavior of StockChangeStatusSerializer
512 if old_custom_status != custom_status:
513 deltas['status'] = custom_status
514 deltas['status_logical'] = self.status
515 else:
516 deltas['status'] = self.status
517 deltas['status_logical'] = self.status
519 if old_custom_status:
520 deltas['old_status'] = old_custom_status
521 deltas['old_status_logical'] = old.status
522 else:
523 deltas['old_status'] = old.status
524 deltas['old_status_logical'] = old.status
526 if add_note and len(deltas) > 0:
527 self.add_tracking_entry(
528 StockHistoryCode.EDITED, user, deltas=deltas, notes=notes
529 )
531 except (ValueError, StockItem.DoesNotExist):
532 pass
534 super().save(*args, **kwargs)
536 # If user information is provided, and no existing note exists, create one!
537 if add_note and self.tracking_info.count() == 0:
538 tracking_info = {'status': self.status}
540 self.add_tracking_entry(
541 StockHistoryCode.CREATED,
542 user,
543 deltas=tracking_info,
544 notes=notes,
545 location=self.location,
546 quantity=float(self.quantity),
547 )
549 def delete(self, ignore_serial_check: bool = False, **kwargs):
550 """Custom delete method for StockItem model.
552 Arguments:
553 ignore_serial_check: If True, allow deletion of serialized stock items regardless of global setting
554 """
555 if not ignore_serial_check and not get_global_setting(
556 'STOCK_ALLOW_DELETE_SERIALIZED', cache=False
557 ):
558 if self.serialized:
559 raise ValidationError(_('Serialized stock items cannot be deleted'))
561 super().delete(**kwargs)
563 @staticmethod
564 def get_api_url():
565 """Return API url."""
566 return reverse('api-stock-list')
568 def api_instance_filters(self):
569 """Custom API instance filters."""
570 return {'parent': {'exclude_tree': self.pk}}
572 @classmethod
573 def barcode_model_type_code(cls):
574 """Return the associated barcode model type code for this model."""
575 return 'SI'
577 def get_test_keys(self, include_installed=True):
578 """Construct a flattened list of test 'keys' for this StockItem."""
579 keys = []
581 for test in self.part.getTestTemplates(required=True):
582 if test.key not in keys:
583 keys.append(test.key)
585 for test in self.part.getTestTemplates(required=False):
586 if test.key not in keys:
587 keys.append(test.key)
589 for result in self.testResultList(include_installed=include_installed):
590 if result.key not in keys:
591 keys.append(result.key)
593 return list(keys)
595 def report_context(self) -> StockItemReportContext:
596 """Generate custom report context data for this StockItem."""
597 return {
598 'barcode_data': self.barcode_data,
599 'barcode_hash': self.barcode_hash,
600 'batch': self.batch,
601 'child_items': self.get_children(),
602 'ipn': self.part.IPN,
603 'installed_items': self.get_installed_items(cascade=True),
604 'item': self,
605 'name': self.part.full_name,
606 'part': self.part,
607 'qr_data': self.barcode,
608 'qr_url': self.get_absolute_url(),
609 'parameters': self.part.parameters_map(),
610 'quantity': InvenTree.helpers.normalize(self.quantity),
611 'result_list': self.testResultList(include_installed=True),
612 'results': self.testResultMap(include_installed=True, cascade=True),
613 'serial': self.serial,
614 'stock_item': self,
615 'tests': self.testResultMap(),
616 'test_keys': self.get_test_keys(),
617 'test_template_list': self.part.getTestTemplates(),
618 'test_templates': self.part.getTestTemplateMap(),
619 }
621 # A Query filter which will be reused in multiple places to determine if a StockItem is actually "in stock"
622 # See also: StockItem.in_stock() method
623 IN_STOCK_FILTER = Q(
624 quantity__gt=0,
625 sales_order=None,
626 belongs_to=None,
627 customer=None,
628 consumed_by=None,
629 is_building=False,
630 status__in=StockStatusGroups.AVAILABLE_CODES,
631 )
633 @classmethod
634 def get_expired_filter(cls):
635 """A query filter which can be used to filter StockItem objects which have expired."""
636 return (
637 cls.IN_STOCK_FILTER
638 & ~Q(expiry_date=None)
639 & Q(expiry_date__lt=InvenTree.helpers.current_date())
640 )
642 @classmethod
643 def _create_serial_numbers(cls, serials: list, **kwargs) -> QuerySet:
644 """Create multiple stock items with the provided serial numbers.
646 Arguments:
647 serials: List of serial numbers to create
648 **kwargs: Additional keyword arguments to pass to the StockItem creation function
650 Returns:
651 QuerySet: The created StockItem objects
653 Raises:
654 ValidationError: If any of the provided serial numbers are invalid
656 This method uses bulk_create to create multiple StockItem objects in a single query,
657 which is much more efficient than creating them one-by-one.
659 However, it does not perform any validation checks on the provided serial numbers,
660 and also does not generate any "stock tracking entries".
662 Note: This is an 'internal' function and should not be used by external code / plugins.
663 """
664 # Ensure the primary-key field is not provided
665 kwargs.pop('id', None)
666 kwargs.pop('pk', None)
668 # Create a list of StockItem objects
669 items = []
671 # Provide some default field values
672 data = {**kwargs}
674 # Extract foreign-key fields from the provided data
675 fk_relations = {
676 'parent': StockItem,
677 'part': PartModels.Part,
678 'build': build.models.Build,
679 'purchase_order': order.models.PurchaseOrder,
680 'supplier_part': CompanyModels.SupplierPart,
681 'location': StockLocation,
682 'belongs_to': StockItem,
683 'customer': CompanyModels.Company,
684 'consumed_by': build.models.Build,
685 'sales_order': order.models.SalesOrder,
686 }
688 for field, model in fk_relations.items():
689 if instance_id := data.pop(f'{field}_id', None):
690 try:
691 instance = model.objects.get(pk=instance_id)
692 data[field] = instance
693 except (ValueError, model.DoesNotExist):
694 raise ValidationError({field: _(f'{field} does not exist')})
696 # Remove some fields which we do not want copied across
697 for field in [
698 'barcode_data',
699 'barcode_hash',
700 'stocktake_date',
701 'stocktake_user',
702 'stocktake_user_id',
703 ]:
704 data.pop(field, None)
706 if 'part' not in data:
707 raise ValidationError({'part': _('Part must be specified')})
709 part = data['part']
711 parent = kwargs.pop('parent', None) or data.get('parent')
712 tree_id = kwargs.pop('tree_id', StockItem.getNextTreeID())
714 if parent:
715 # Override with parent's tree_id if provided
716 tree_id = parent.tree_id
718 # Pre-calculate MPTT fields
719 data['parent'] = parent if parent else None
720 data['level'] = parent.level + 1 if parent else 0
721 data['lft'] = 0 if parent else 1
722 data['rght'] = 0 if parent else 2
724 # Force single quantity for each item
725 data['quantity'] = 1
727 for serial in serials:
728 data['serial'] = serial
730 if serial is not None:
731 data['serial_int'] = StockItem.convert_serial_to_int(serial) or 0
732 else:
733 data['serial_int'] = 0
735 data['tree_id'] = tree_id
737 if not parent:
738 # No parent, this is a top-level item, so increment the tree_id
739 # This is because each new item is a "top-level" node in the StockItem tree
740 tree_id += 1
742 # Construct a new StockItem from the provided dict
743 items.append(StockItem(**data))
745 # Create the StockItem objects in bulk
746 StockItem.objects.bulk_create(items, batch_size=250)
748 # We will need to rebuild the stock item tree manually, due to the bulk_create operation
749 if parent and parent.tree_id:
750 # Rebuild the tree structure for this StockItem tree
751 logger.info(
752 'Rebuilding StockItem tree structure for tree_id: %s', parent.tree_id
753 )
754 stock.tasks.rebuild_stock_item_tree(parent.tree_id)
756 # Fetch the new StockItem objects from the database
757 items = StockItem.objects.filter(part=part, serial__in=serials)
759 # Trigger a 'created' event for the new items
760 # Note that instead of a single event for each item,
761 # we trigger a single event for all items created
762 stock_ids = list(items.values_list('id', flat=True).distinct())
763 trigger_event(StockEvents.ITEMS_CREATED, ids=stock_ids)
765 # Return the newly created StockItem objects
766 return items
768 @staticmethod
769 def convert_serial_to_int(serial: str) -> int | None:
770 """Convert the provided serial number to an integer value.
772 This function hooks into the plugin system to allow for custom serial number conversion.
773 """
774 from plugin import PluginMixinEnum, registry
776 # First, let any plugins convert this serial number to an integer value
777 # If a non-null value is returned (by any plugin) we will use that
778 if not InvenTree.ready.isReadOnlyCommand():
779 for plugin in registry.with_mixin(PluginMixinEnum.VALIDATION):
780 try:
781 serial_int = plugin.convert_serial_to_int(serial)
782 except Exception:
783 InvenTree.exceptions.log_error(
784 'convert_serial_to_int', plugin=plugin.slug
785 )
786 serial_int = None
788 # Save the first returned result
789 if serial_int is not None:
790 # Ensure that it is clipped within a range allowed in the database schema
791 clip = 0x7FFFFFFF
792 serial_int = abs(serial_int)
793 serial_int = min(serial_int, clip)
794 # Return the first non-null value
795 return serial_int
797 # None of the plugins provided a valid integer value
798 if serial not in [None, '']:
799 return InvenTree.helpers.extract_int(serial)
800 else:
801 return None
803 def update_serial_number(self):
804 """Update the 'serial_int' field, to be an integer representation of the serial number.
806 This is used for efficient numerical sorting
807 """
808 serial = str(getattr(self, 'serial', '')).strip()
810 if not serial:
811 self.serial_int = 0
812 return
814 serial_int = self.convert_serial_to_int(serial)
816 try:
817 serial_int = int(serial_int)
819 if serial_int <= 0:
820 serial_int = 0
821 except (ValueError, TypeError):
822 serial_int = 0
824 self.serial_int = serial_int
826 def get_next_serialized_item(self, include_variants=True, reverse=False):
827 """Get the "next" serial number for the part this stock item references.
829 e.g. if this stock item has a serial number 100, we may return the stock item with serial number 101
831 Note that this only works for "serialized" stock items with integer values
833 Args:
834 include_variants: True if we wish to include stock for variant parts
835 reverse: True if we want to return the "previous" (lower) serial number
837 Returns:
838 A StockItem object matching the requirements, or None
839 """
840 if not self.serialized:
841 return None
843 # Find only serialized stock items
844 items = StockItem.objects.exclude(serial=None).exclude(serial='')
846 if include_variants:
847 # Match against any part within the variant tree
848 items = items.filter(part__tree_id=self.part.tree_id)
849 else:
850 # Match only against the specific part
851 items = items.filter(part=self.part)
853 serial = self.serial_int
855 if reverse:
856 # Select only stock items with lower serial numbers, in decreasing order
857 items = items.filter(serial_int__lt=serial)
858 items = items.order_by('-serial_int')
859 else:
860 # Select only stock items with higher serial numbers, in increasing order
861 items = items.filter(serial_int__gt=serial)
862 items = items.order_by('serial_int')
864 if items.count() > 0:
865 item = items.first()
867 if item.serialized:
868 return item
870 return None
872 @property
873 def get_next_stock_item(self):
874 """Return the 'next' stock item (based on serial number)."""
875 return self.get_next_serialized_item()
877 @property
878 def get_previous_stock_item(self):
879 """Return the 'previous' stock item (based on serial number)."""
880 return self.get_next_serialized_item(reverse=True)
882 @property
883 def status_label(self):
884 """Return label."""
885 return StockStatus.label(self.status)
887 @property
888 def serialized(self):
889 """Return True if this StockItem is serialized."""
890 return (
891 self.serial is not None
892 and len(str(self.serial).strip()) > 0
893 and self.quantity == 1
894 )
896 def validate_unique(self, exclude=None):
897 """Test that this StockItem is "unique".
899 If the StockItem is serialized, the same serial number.
900 cannot exist for the same part (or part tree).
901 """
902 super().validate_unique(exclude)
904 # If the serial number is set, make sure it is not a duplicate
905 if self.serial:
906 self.serial = str(self.serial).strip()
908 try:
909 self.part.validate_serial_number(self.serial, self, raise_error=True)
910 except ValidationError as exc:
911 raise ValidationError({'serial': exc.message})
913 def validate_batch_code(self):
914 """Ensure that the batch code is valid for this StockItem.
916 - Validation is performed by custom plugins.
917 - By default, no validation checks are performed
918 """
919 from plugin import PluginMixinEnum, registry
921 if not InvenTree.ready.isReadOnlyCommand():
922 for plugin in registry.with_mixin(PluginMixinEnum.VALIDATION):
923 try:
924 plugin.validate_batch_code(self.batch, self)
925 except ValidationError as exc:
926 raise ValidationError({'batch': exc.message})
927 except Exception:
928 InvenTree.exceptions.log_error(
929 'validate_batch_code', plugin=plugin.slug
930 )
932 def clean(self):
933 """Validate the StockItem object (separate to field validation).
935 The following validation checks are performed:
936 - The 'part' and 'supplier_part.part' fields cannot point to the same Part object
937 - The 'part' is not virtual
938 - The 'part' does not belong to itself
939 - The location is not structural
940 - Quantity must be 1 if the StockItem has a serial number
941 """
942 if self.location is not None and self.location.structural:
943 raise ValidationError({
944 'location': _(
945 'Stock items cannot be located into structural stock locations!'
946 )
947 })
949 super().clean()
951 # Strip serial number field
952 if type(self.serial) is str:
953 self.serial = self.serial.strip()
955 # Strip batch code field
956 if type(self.batch) is str:
957 self.batch = self.batch.strip()
959 if not get_global_setting('STOCK_ALLOW_EDIT_SERIAL'):
960 deltas = self.get_field_deltas()
962 # Prevent editing of serial numbers if the item already has a serial number assigned
963 if 'serial' in deltas and deltas['serial']['old'] not in [None, '']:
964 raise ValidationError({
965 'serial': _(
966 'Editing of serial numbers is not allowed - this item has already been assigned a serial number'
967 )
968 })
970 # Custom validation of batch code
971 self.validate_batch_code()
973 try:
974 # Trackable parts must have integer values for quantity field!
975 if self.part.trackable and self.quantity != int(self.quantity):
976 raise ValidationError({
977 'quantity': _('Quantity must be integer value for trackable parts')
978 })
980 # Virtual parts cannot have stock items created against them
981 if self.part.virtual:
982 raise ValidationError({
983 'part': _('Stock item cannot be created for virtual parts')
984 })
985 except PartModels.Part.DoesNotExist:
986 # For some reason the 'clean' process sometimes throws errors because self.part does not exist
987 # It *seems* that this only occurs in unit testing, though.
988 # Probably should investigate this at some point.
989 pass
991 if self.quantity < 0:
992 raise ValidationError({'quantity': _('Quantity must be greater than zero')})
994 # The 'supplier_part' field must point to the same part!
995 try:
996 if self.supplier_part is not None:
997 if self.supplier_part.part != self.part:
998 raise ValidationError({
999 'supplier_part': _(
1000 f"Part type ('{self.supplier_part.part}') must be {self.part}"
1001 )
1002 })
1004 if self.part is not None:
1005 # A part with a serial number MUST have the quantity set to 1
1006 if self.serial:
1007 if self.quantity > 1:
1008 raise ValidationError({
1009 'quantity': _(
1010 'Quantity must be 1 for item with a serial number'
1011 ),
1012 'serial': _(
1013 'Serial number cannot be set if quantity greater than 1'
1014 ),
1015 })
1017 if self.quantity == 0:
1018 self.quantity = 1
1020 elif self.quantity > 1:
1021 raise ValidationError({
1022 'quantity': _(
1023 'Quantity must be 1 for item with a serial number'
1024 )
1025 })
1027 # Serial numbered items cannot be deleted on depletion
1028 self.delete_on_deplete = False
1030 except PartModels.Part.DoesNotExist:
1031 pass
1033 # Ensure that the item cannot be assigned to itself
1034 if self.belongs_to and self.belongs_to.pk == self.pk:
1035 raise ValidationError({'belongs_to': _('Item cannot belong to itself')})
1037 # If the item is marked as "is_building", it must point to a build!
1038 if self.is_building and not self.build:
1039 raise ValidationError({
1040 'build': _('Item must have a build reference if is_building=True')
1041 })
1043 # If the item points to a build, check that the Part references match
1044 if self.build:
1045 if self.part == self.build.part:
1046 # Part references match exactly
1047 pass
1048 elif self.part in self.build.part.get_conversion_options():
1049 # Part reference is one of the valid conversion options for the build output
1050 pass
1051 else:
1052 raise ValidationError({
1053 'build': _('Build reference does not point to the same part object')
1054 })
1056 def get_absolute_url(self):
1057 """Return url for instance."""
1058 return InvenTree.helpers.pui_url(f'/stock/item/{self.id}')
1060 def get_part_name(self):
1061 """Returns part name."""
1062 return self.part.full_name
1064 # Note: When a StockItem is deleted, a pre_delete signal handles the parent/child relationship
1065 parent = TreeForeignKey(
1066 'self',
1067 verbose_name=_('Parent Stock Item'),
1068 on_delete=models.DO_NOTHING,
1069 blank=True,
1070 null=True,
1071 related_name='children',
1072 )
1074 part = models.ForeignKey(
1075 'part.Part',
1076 on_delete=models.CASCADE,
1077 verbose_name=_('Base Part'),
1078 related_name='stock_items',
1079 help_text=_('Base part'),
1080 limit_choices_to={'virtual': False},
1081 )
1083 supplier_part = models.ForeignKey(
1084 'company.SupplierPart',
1085 blank=True,
1086 null=True,
1087 on_delete=models.SET_NULL,
1088 verbose_name=_('Supplier Part'),
1089 help_text=_('Select a matching supplier part for this stock item'),
1090 related_name='stock_items',
1091 )
1093 # Note: When a StockLocation is deleted, stock items are updated via a signal
1094 location = TreeForeignKey(
1095 StockLocation,
1096 on_delete=models.DO_NOTHING,
1097 verbose_name=_('Stock Location'),
1098 related_name='stock_items',
1099 blank=True,
1100 null=True,
1101 help_text=_('Where is this stock item located?'),
1102 )
1104 packaging = models.CharField(
1105 max_length=50,
1106 blank=True,
1107 null=True,
1108 verbose_name=_('Packaging'),
1109 help_text=_('Packaging this stock item is stored in'),
1110 )
1112 # When deleting a stock item with installed items, those installed items are also installed
1113 belongs_to = models.ForeignKey(
1114 'self',
1115 verbose_name=_('Installed In'),
1116 on_delete=models.CASCADE,
1117 related_name='installed_parts',
1118 blank=True,
1119 null=True,
1120 help_text=_('Is this item installed in another item?'),
1121 )
1123 customer = models.ForeignKey(
1124 CompanyModels.Company,
1125 on_delete=models.SET_NULL,
1126 null=True,
1127 blank=True,
1128 limit_choices_to={'is_customer': True},
1129 related_name='assigned_stock',
1130 help_text=_('Customer'),
1131 verbose_name=_('Customer'),
1132 )
1134 serial = models.CharField(
1135 verbose_name=_('Serial Number'),
1136 max_length=100,
1137 blank=True,
1138 null=True,
1139 help_text=_('Serial number for this item'),
1140 )
1142 serial_int = models.IntegerField(default=0)
1144 link = InvenTreeURLField(
1145 verbose_name=_('External Link'),
1146 blank=True,
1147 help_text=_('Link to external URL'),
1148 max_length=2000,
1149 )
1151 batch = models.CharField(
1152 verbose_name=_('Batch Code'),
1153 max_length=100,
1154 blank=True,
1155 null=True,
1156 help_text=_('Batch code for this stock item'),
1157 default=generate_batch_code,
1158 )
1160 quantity = models.DecimalField(
1161 verbose_name=_('Stock Quantity'),
1162 max_digits=15,
1163 decimal_places=5,
1164 validators=[MinValueValidator(0)],
1165 default=1,
1166 )
1168 build = models.ForeignKey(
1169 'build.Build',
1170 on_delete=models.SET_NULL,
1171 verbose_name=_('Source Build'),
1172 blank=True,
1173 null=True,
1174 help_text=_('Build for this stock item'),
1175 related_name='build_outputs',
1176 )
1178 consumed_by = models.ForeignKey(
1179 'build.Build',
1180 on_delete=models.CASCADE,
1181 verbose_name=_('Consumed By'),
1182 blank=True,
1183 null=True,
1184 help_text=_('Build order which consumed this stock item'),
1185 related_name='consumed_stock',
1186 )
1188 is_building = models.BooleanField(default=False)
1190 purchase_order = models.ForeignKey(
1191 'order.PurchaseOrder',
1192 on_delete=models.SET_NULL,
1193 verbose_name=_('Source Purchase Order'),
1194 related_name='stock_items',
1195 blank=True,
1196 null=True,
1197 help_text=_('Purchase order for this stock item'),
1198 )
1200 sales_order = models.ForeignKey(
1201 'order.SalesOrder',
1202 on_delete=models.SET_NULL,
1203 verbose_name=_('Destination Sales Order'),
1204 related_name='stock_items',
1205 null=True,
1206 blank=True,
1207 )
1209 expiry_date = models.DateField(
1210 blank=True,
1211 null=True,
1212 verbose_name=_('Expiry Date'),
1213 help_text=_(
1214 'Expiry date for stock item. Stock will be considered expired after this date'
1215 ),
1216 )
1218 stocktake_date = models.DateField(blank=True, null=True)
1220 stocktake_user = models.ForeignKey(
1221 User,
1222 on_delete=models.SET_NULL,
1223 blank=True,
1224 null=True,
1225 related_name='stocktake_stock',
1226 )
1228 creation_date = models.DateTimeField(
1229 null=True,
1230 blank=True,
1231 auto_now_add=True,
1232 editable=False,
1233 verbose_name=_('Creation Date'),
1234 help_text=_('Date that this stock item was created'),
1235 )
1237 delete_on_deplete = models.BooleanField(
1238 default=default_delete_on_deplete,
1239 verbose_name=_('Delete on deplete'),
1240 help_text=_('Delete this Stock Item when stock is depleted'),
1241 )
1243 status = InvenTreeCustomStatusModelField(
1244 default=StockStatus.OK.value,
1245 status_class=StockStatus,
1246 choices=StockStatus.items(),
1247 validators=[MinValueValidator(0)],
1248 )
1250 @property
1251 def status_text(self):
1252 """Return the text representation of the status field."""
1253 return StockStatus.text(self.status)
1255 purchase_price = InvenTreeModelMoneyField(
1256 max_digits=19,
1257 decimal_places=6,
1258 blank=True,
1259 null=True,
1260 verbose_name=_('Purchase Price'),
1261 help_text=_('Single unit purchase price at time of purchase'),
1262 )
1264 owner = models.ForeignKey(
1265 Owner,
1266 on_delete=models.SET_NULL,
1267 blank=True,
1268 null=True,
1269 verbose_name=_('Owner'),
1270 help_text=_('Select Owner'),
1271 related_name='stock_items',
1272 )
1274 @transaction.atomic
1275 def convert_to_variant(self, variant, user, notes=None):
1276 """Convert this StockItem instance to a "variant", i.e. change the "part" reference field."""
1277 if not variant:
1278 # Ignore null values
1279 return
1281 if variant == self.part:
1282 # Variant is the same as the current part
1283 return
1285 self.part = variant
1286 self.save()
1288 self.add_tracking_entry(
1289 StockHistoryCode.CONVERTED_TO_VARIANT,
1290 user,
1291 deltas={'part': variant.pk},
1292 notes=_('Converted to part') + ': ' + variant.full_name,
1293 )
1295 def get_item_owner(self):
1296 """Return the closest "owner" for this StockItem.
1298 - If the item has an owner set, return that
1299 - If the item is "in stock", check the StockLocation
1300 - Otherwise, return None
1301 """
1302 if self.owner is not None:
1303 return self.owner
1305 if self.in_stock and self.location is not None:
1306 loc_owner = self.location.get_location_owner()
1308 if loc_owner:
1309 return loc_owner
1311 return None
1313 def check_ownership(self, user):
1314 """Check if the user "owns" (or is one of the owners of) the item."""
1315 # Superuser accounts automatically "own" everything
1316 if user.is_superuser:
1317 return True
1319 ownership_enabled = get_global_setting('STOCK_OWNERSHIP_CONTROL')
1321 if not ownership_enabled:
1322 # Location ownership function is not enabled, so return True
1323 return True
1325 owner = self.get_item_owner()
1327 if owner is None:
1328 return True
1330 return owner.is_user_allowed(user, include_group=True)
1332 def is_stale(self):
1333 """Returns True if this Stock item is "stale".
1335 To be "stale", the following conditions must be met:
1336 - Expiry date is not None
1337 - Expiry date will "expire" within the configured stale date
1338 - The StockItem is otherwise "in stock"
1339 """
1340 if self.expiry_date is None:
1341 return False
1343 if not self.in_stock:
1344 return False
1346 today = InvenTree.helpers.current_date()
1348 stale_days = get_global_setting('STOCK_STALE_DAYS')
1350 if stale_days <= 0:
1351 return False
1353 expiry_date = today + timedelta(days=stale_days)
1355 return self.expiry_date < expiry_date
1357 def is_expired(self):
1358 """Returns True if this StockItem is "expired".
1360 To be "expired", the following conditions must be met:
1361 - Expiry date is not None
1362 - Expiry date is "in the past"
1363 - The StockItem is otherwise "in stock"
1364 """
1365 if self.expiry_date is None:
1366 return False
1368 if not self.in_stock:
1369 return False
1371 today = InvenTree.helpers.current_date()
1373 return self.expiry_date < today
1375 def clearAllocations(self):
1376 """Clear all order allocations for this StockItem.
1378 Clears:
1379 - SalesOrder allocations
1380 - Build allocations
1381 """
1382 # Delete outstanding SalesOrder allocations
1383 self.sales_order_allocations.all().delete()
1385 # Delete outstanding BuildOrder allocations
1386 self.allocations.all().delete()
1388 def allocateToCustomer(
1389 self, customer, quantity=None, order=None, user=None, notes=None
1390 ):
1391 """Allocate a StockItem to a customer.
1393 This action can be called by the following processes:
1394 - Completion of a SalesOrder
1395 - User manually assigns a StockItem to the customer
1397 Args:
1398 customer: The customer (Company) to assign the stock to
1399 quantity: Quantity to assign (if not supplied, total quantity is used)
1400 order: SalesOrder reference
1401 user: User that performed the action
1402 notes: Notes field
1403 """
1404 if quantity is None:
1405 quantity = self.quantity
1407 if quantity >= self.quantity:
1408 item = self
1409 else:
1410 item = self.splitStock(quantity, None, user)
1412 # Update StockItem fields with new information
1413 item.sales_order = order
1414 item.customer = customer
1415 item.location = None
1417 item.save(add_note=False)
1419 code = StockHistoryCode.SENT_TO_CUSTOMER
1420 deltas = {'quantity': float(quantity)}
1422 if customer is not None:
1423 deltas['customer'] = customer.pk
1424 deltas['customer_name'] = customer.name
1426 # If an order is provided, we are shipping against a SalesOrder, not manually!
1427 if order:
1428 code = StockHistoryCode.SHIPPED_AGAINST_SALES_ORDER
1429 deltas['salesorder'] = order.pk
1431 item.add_tracking_entry(code, user, deltas, notes=notes)
1433 trigger_event(
1434 StockEvents.ITEM_ASSIGNED_TO_CUSTOMER,
1435 id=self.id,
1436 customer=customer.id if customer else None,
1437 )
1439 # Return the reference to the stock item
1440 return item
1442 @transaction.atomic
1443 def return_from_customer(self, location, user=None, **kwargs):
1444 """Return stock item from customer, back into the specified location.
1446 Arguments:
1447 location: The location to return the stock item to
1448 user: The user performing the action
1450 Keyword Arguments:
1451 notes: Additional notes to add to the tracking entry
1452 status: Optionally set the status of the stock item
1454 If the selected location is the same as the parent, merge stock back into the parent.
1455 Otherwise create the stock in the new location.
1457 Note that this function is provided for legacy compatibility,
1458 and the 'return_to_stock' function should be used instead.
1459 """
1460 self.return_to_stock(
1461 location,
1462 user,
1463 tracking_code=StockHistoryCode.RETURNED_FROM_CUSTOMER,
1464 **kwargs,
1465 )
1467 @transaction.atomic
1468 def return_to_stock(
1469 self, location, user=None, quantity=None, merge: bool = True, **kwargs
1470 ):
1471 """Return stock item into stock, removing any consumption status.
1473 Arguments:
1474 location: The location to return the stock item to
1475 user: The user performing the action
1476 quantity: If specified, the quantity to return to stock (default is the full quantity)
1477 merge: If True, attempt to merge this stock item back into the parent stock item
1478 """
1479 notes = kwargs.get('notes', '')
1481 tracking_code = kwargs.get('tracking_code', StockHistoryCode.RETURNED_TO_STOCK)
1483 item = self
1485 if quantity is not None and not self.serialized:
1486 # If quantity is specified, we are splitting the stock item
1487 if quantity <= 0:
1488 raise ValidationError({
1489 'quantity': _('Quantity must be greater than zero')
1490 })
1492 if quantity > self.quantity:
1493 raise ValidationError({
1494 'quantity': _('Quantity exceeds available stock')
1495 })
1497 if quantity < self.quantity:
1498 # Split the stock item
1499 item = self.splitStock(quantity, None, user)
1501 tracking_info = {
1502 'quantity': float(quantity)
1503 if quantity is not None
1504 else float(item.quantity)
1505 }
1507 if location:
1508 tracking_info['location'] = location.pk
1510 if item.customer:
1511 tracking_info['customer'] = item.customer.id
1512 tracking_info['customer_name'] = item.customer.name
1514 if item.consumed_by:
1515 tracking_info['build_order'] = item.consumed_by.id
1517 # Clear out allocation information for the stock item
1518 item.consumed_by = None
1519 item.customer = None
1520 item.belongs_to = None
1521 item.sales_order = None
1522 item.location = location
1524 if status := kwargs.pop('status', None):
1525 if not item.compare_status(status):
1526 old_custom_status = item.get_custom_status()
1527 old_status_logical = item.status
1528 item.set_status(status)
1529 tracking_info['status'] = status # may be a custom value
1530 tracking_info['status_logical'] = (
1531 item.status
1532 ) # always the logical value
1533 tracking_info['old_status'] = (
1534 old_custom_status if old_custom_status else old_status_logical
1535 )
1536 tracking_info['old_status_logical'] = old_status_logical
1538 item.save()
1540 item.clearAllocations()
1542 item.add_tracking_entry(
1543 tracking_code, user, notes=notes, deltas=tracking_info, location=location
1544 )
1546 trigger_event(StockEvents.ITEM_RETURNED_TO_STOCK, id=item.id)
1548 # Attempt to merge returned item into parent item:
1549 # - 'merge' parameter is True
1550 # - The parent location is the same as the current location
1551 # - The item does not have a serial number
1553 if (
1554 merge
1555 and not item.serialized
1556 and self.parent
1557 and item.location == self.parent.location
1558 ):
1559 self.parent.merge_stock_items(
1560 {item}, user=user, location=location, notes=notes
1561 )
1562 else:
1563 item.save(add_note=False)
1565 def is_allocated(self):
1566 """Return True if this StockItem is allocated to a SalesOrder, TransferOrder, or a Build."""
1567 return self.allocation_count() > 0
1569 def build_allocation_count(self, **kwargs):
1570 """Return the total quantity allocated to builds, with optional filters."""
1571 query = self.allocations.all()
1573 if filter_allocations := kwargs.get('filter_allocations'):
1574 query = query.filter(**filter_allocations)
1576 if exclude_allocations := kwargs.get('exclude_allocations'):
1577 query = query.exclude(**exclude_allocations)
1579 query = query.aggregate(q=Coalesce(Sum('quantity'), Decimal(0)))
1581 total = query['q']
1583 if total is None:
1584 total = Decimal(0)
1586 return total
1588 def get_sales_order_allocations(self, active=True, **kwargs):
1589 """Return a queryset for SalesOrderAllocations against this StockItem, with optional filters.
1591 Arguments:
1592 active: Filter by 'active' status of the allocation
1593 """
1594 query = self.sales_order_allocations.all()
1596 if filter_allocations := kwargs.get('filter_allocations'):
1597 query = query.filter(**filter_allocations)
1599 if exclude_allocations := kwargs.get('exclude_allocations'):
1600 query = query.exclude(**exclude_allocations)
1602 if active is True:
1603 query = query.filter(
1604 line__order__status__in=SalesOrderStatusGroups.OPEN,
1605 shipment__shipment_date=None,
1606 )
1607 elif active is False:
1608 query = query.exclude(
1609 line__order__status__in=SalesOrderStatusGroups.OPEN
1610 ).exclude(shipment__shipment_date=None)
1612 return query
1614 def sales_order_allocation_count(self, active=True, **kwargs):
1615 """Return the total quantity allocated to SalesOrders."""
1616 query = self.get_sales_order_allocations(active=active, **kwargs)
1617 query = query.aggregate(q=Coalesce(Sum('quantity'), Decimal(0)))
1619 total = query['q']
1621 if total is None:
1622 total = Decimal(0)
1624 return total
1626 def get_transfer_order_allocations(self, active=True, **kwargs):
1627 """Return a queryset for TransferOrderAllocations against this StockItem, with optional filters.
1629 Arguments:
1630 active: Filter by 'active' status of the allocation
1631 """
1632 query = self.transfer_order_allocations.all()
1634 if filter_allocations := kwargs.get('filter_allocations'):
1635 query = query.filter(**filter_allocations)
1637 if exclude_allocations := kwargs.get('exclude_allocations'):
1638 query = query.exclude(**exclude_allocations)
1640 if active is True:
1641 query = query.filter(line__order__status__in=TransferOrderStatusGroups.OPEN)
1642 elif active is False:
1643 query = query.exclude(
1644 line__order__status__in=TransferOrderStatusGroups.OPEN
1645 )
1647 return query
1649 def transfer_order_allocation_count(self, active=True, **kwargs):
1650 """Return the total quantity allocated to TransferOrders."""
1651 query = self.get_transfer_order_allocations(active=active, **kwargs)
1652 query = query.aggregate(q=Coalesce(Sum('quantity'), Decimal(0)))
1654 total = query['q']
1656 if total is None:
1657 total = Decimal(0)
1659 return total
1661 def allocation_count(self):
1662 """Return the total quantity allocated to builds or orders."""
1663 bo = self.build_allocation_count()
1664 so = self.sales_order_allocation_count()
1665 to = self.transfer_order_allocation_count()
1667 return bo + so + to
1669 def unallocated_quantity(self):
1670 """Return the quantity of this StockItem which is *not* allocated."""
1671 return max(self.quantity - self.allocation_count(), 0)
1673 def can_delete(self):
1674 """Can this stock item be deleted?
1676 It can NOT be deleted under the following circumstances:
1677 - Has installed stock items
1678 - Is installed inside another StockItem
1679 - It has been assigned to a SalesOrder
1680 - It has been assigned to a BuildOrder
1681 """
1682 if self.installed_item_count() > 0:
1683 return False
1685 return self.sales_order is None
1687 def get_installed_items(self, cascade: bool = False) -> set[StockItem]:
1688 """Return all stock items which are *installed* in this one!
1690 Note: This function is recursive, and may result in a number of database hits!
1692 Args:
1693 cascade (bool, optional): Include items which are installed in items which are installed in items. Defaults to False.
1695 Returns:
1696 set[StockItem]: Sll stock items which are installed
1697 """
1698 installed = set()
1700 items = StockItem.objects.filter(belongs_to=self)
1702 for item in items:
1703 # Prevent duplication or recursion
1704 if item == self or item in installed:
1705 continue
1707 installed.add(item)
1709 if cascade:
1710 sub_items = item.get_installed_items(cascade=True)
1712 for sub_item in sub_items:
1713 # Prevent recursion
1714 if sub_item == self or sub_item in installed:
1715 continue
1717 installed.add(sub_item)
1719 return installed
1721 def installed_item_count(self):
1722 """Return the number of stock items installed inside this one."""
1723 return self.installed_parts.count()
1725 @transaction.atomic
1726 def installStockItem(self, other_item, quantity, user, notes, build=None):
1727 """Install another stock item into this stock item.
1729 Args:
1730 other_item: The stock item to install into this stock item
1731 quantity: The quantity of stock to install
1732 user: The user performing the operation
1733 notes: Any notes associated with the operation
1734 build: The BuildOrder to associate with the operation (optional)
1735 """
1736 # If the quantity is less than the stock item, split the stock!
1737 stock_item = other_item.splitStock(quantity, None, user)
1739 if stock_item is None:
1740 stock_item = other_item
1742 # Assign the other stock item into this one
1743 stock_item.belongs_to = self
1745 if build is not None:
1746 stock_item.consumed_by = build
1748 stock_item.location = None
1749 stock_item.save(add_note=False)
1751 deltas = {'stockitem': self.pk, 'quantity': float(quantity)}
1753 if build is not None:
1754 deltas['buildorder'] = build.pk
1756 # Add a transaction note to the other item
1757 stock_item.add_tracking_entry(
1758 StockHistoryCode.INSTALLED_INTO_ASSEMBLY, user, notes=notes, deltas=deltas
1759 )
1761 # Add a transaction note to this item (the assembly)
1762 self.add_tracking_entry(
1763 StockHistoryCode.INSTALLED_CHILD_ITEM,
1764 user,
1765 notes=notes,
1766 deltas={'stockitem': stock_item.pk, 'quantity': float(quantity)},
1767 )
1769 trigger_event(
1770 StockEvents.ITEM_INSTALLED_INTO_ASSEMBLY,
1771 id=stock_item.pk,
1772 assembly_id=self.pk,
1773 )
1775 @transaction.atomic
1776 def uninstall_into_location(self, location, user, notes):
1777 """Uninstall this stock item from another item, into a location.
1779 Args:
1780 location: The stock location where the item will be moved
1781 user: The user performing the operation
1782 notes: Any notes associated with the operation
1783 """
1784 # If the stock item is not installed in anything, ignore
1785 if self.belongs_to is None:
1786 return False
1788 # Add a transaction note to the parent item
1789 self.belongs_to.add_tracking_entry(
1790 StockHistoryCode.REMOVED_CHILD_ITEM,
1791 user,
1792 deltas={'stockitem': self.pk, 'quantity': float(self.quantity)},
1793 notes=notes,
1794 )
1796 tracking_info = {
1797 'stockitem': self.belongs_to.pk,
1798 'quantity': float(self.quantity),
1799 }
1801 self.add_tracking_entry(
1802 StockHistoryCode.REMOVED_FROM_ASSEMBLY,
1803 user,
1804 notes=notes,
1805 deltas=tracking_info,
1806 location=location,
1807 )
1809 # Mark this stock item as *not* belonging to anyone
1810 self.belongs_to = None
1811 self.consumed_by = None
1812 self.location = location
1814 self.save()
1816 @property
1817 def children(self):
1818 """Return a list of the child items which have been split from this stock item."""
1819 return self.get_descendants(include_self=False)
1821 @property
1822 def child_count(self):
1823 """Return the number of 'child' items associated with this StockItem.
1825 A child item is one which has been split from this one.
1826 """
1827 return self.children.count()
1829 def is_in_stock(
1830 self,
1831 check_status: bool = True,
1832 check_quantity: bool = True,
1833 check_in_production: bool = True,
1834 ) -> bool:
1835 """Return True if this StockItem is "in stock".
1837 Arguments:
1838 check_status: If True, check the status of the StockItem. Defaults to True.
1839 check_quantity: If True, check the quantity of the StockItem. Defaults to True.
1840 check_in_production: If True, check if the item is in production. Defaults to True.
1841 """
1842 if check_status and self.status not in StockStatusGroups.AVAILABLE_CODES:
1843 return False
1845 if check_quantity and self.quantity <= 0:
1846 return False
1848 if check_in_production and self.is_building:
1849 return False
1851 return all([
1852 self.sales_order is None, # Not assigned to a SalesOrder
1853 self.belongs_to is None, # Not installed inside another StockItem
1854 self.customer is None, # Not assigned to a customer
1855 self.consumed_by is None, # Not consumed by a build
1856 ])
1858 @property
1859 def in_stock(self) -> bool:
1860 """Returns True if this item is in stock.
1862 See also: StockItem.IN_STOCK_FILTER for the db optimized version of this check.
1863 """
1864 return self.is_in_stock(check_status=True)
1866 @property
1867 def can_adjust_location(self):
1868 """Returns True if the stock location can be "adjusted" for this part.
1870 Cannot be adjusted if:
1871 - Has been delivered to a customer
1872 - Has been installed inside another StockItem
1873 """
1874 if self.customer is not None:
1875 return False
1877 if self.belongs_to is not None:
1878 return False
1880 return self.sales_order is None
1882 @property
1883 def tracking_info_count(self):
1884 """How many tracking entries are available?"""
1885 return self.tracking_info.count()
1887 @property
1888 def has_tracking_info(self):
1889 """Is tracking info available?"""
1890 return self.tracking_info_count > 0
1892 def add_tracking_entry(
1893 self,
1894 entry_type: int,
1895 user: User,
1896 deltas: dict | None = None,
1897 notes: str = '',
1898 commit: bool = True,
1899 **kwargs,
1900 ):
1901 """Add a history tracking entry for this StockItem.
1903 Args:
1904 entry_type (int): Code describing the "type" of historical action (see StockHistoryCode)
1905 user (User): The user performing this action
1906 deltas (dict, optional): A map of the changes made to the model. Defaults to None.
1907 notes (str, optional): URL associated with this tracking entry. Defaults to ''.
1908 commit (bool, optional): If True, save the entry to the database. Defaults to True.
1910 Returns:
1911 StockItemTracking: The created tracking entry
1912 """
1913 if deltas is None:
1914 deltas = {}
1916 # Prevent empty entry
1917 if (
1918 entry_type == StockHistoryCode.STOCK_UPDATE
1919 and len(deltas) == 0
1920 and not notes
1921 ):
1922 return None
1924 # Has a location been specified?
1925 location = kwargs.get('location')
1927 if location:
1928 deltas['location'] = location.id
1930 # Quantity specified?
1931 quantity = kwargs.get('quantity')
1933 if quantity:
1934 deltas['quantity'] = float(quantity)
1936 entry = StockItemTracking(
1937 item=self,
1938 part=self.part,
1939 tracking_type=entry_type.value,
1940 user=user,
1941 date=InvenTree.helpers.current_time(),
1942 notes=notes,
1943 deltas=deltas,
1944 )
1946 if commit:
1947 entry.save()
1949 return entry
1951 @transaction.atomic
1952 def serializeStock(
1953 self,
1954 quantity: int,
1955 serials: list[str],
1956 user: User | None = None,
1957 notes: str | None = '',
1958 location: StockLocation | None = None,
1959 ):
1960 """Split this stock item into unique serial numbers.
1962 - Quantity can be less than or equal to the quantity of the stock item
1963 - Number of serial numbers must match the quantity
1964 - Provided serial numbers must not already be in use
1966 Arguments:
1967 quantity: Number of items to serialize (integer)
1968 serials: List of serial numbers
1969 user: User object associated with action
1970 notes: Optional notes for tracking
1971 location: If specified, serialized items will be placed in the given location
1973 Returns:
1974 List of newly created StockItem objects, each with a unique serial number.
1975 """
1976 # Cannot serialize stock that is already serialized!
1977 if self.serialized:
1978 return None
1980 if not self.part.trackable:
1981 raise ValidationError({'part': _('Part is not set as trackable')})
1983 # Quantity must be a valid integer value
1984 try:
1985 quantity = int(quantity)
1986 except ValueError:
1987 raise ValidationError({'quantity': _('Quantity must be integer')})
1989 if quantity <= 0:
1990 raise ValidationError({'quantity': _('Quantity must be greater than zero')})
1992 if quantity > self.quantity:
1993 raise ValidationError({
1994 'quantity': _(
1995 f'Quantity must not exceed available stock quantity ({self.quantity})'
1996 )
1997 })
1999 if type(serials) not in [list, tuple]:
2000 raise ValidationError({
2001 'serial_numbers': _('Serial numbers must be provided as a list')
2002 })
2004 if quantity != len(serials):
2005 raise ValidationError({
2006 'quantity': _('Quantity does not match serial numbers')
2007 })
2009 # Test if each of the serial numbers are valid
2010 existing = self.part.find_conflicting_serial_numbers(serials)
2012 if len(existing) > 0:
2013 msg = _('The following serial numbers already exist or are invalid')
2014 msg += ' : '
2015 msg += ','.join([str(x) for x in existing])
2016 raise ValidationError({'serial_numbers': msg})
2018 # Serialize this StockItem
2019 data = dict(StockItem.objects.filter(pk=self.pk).values()[0])
2021 if location:
2022 if location.structural:
2023 raise ValidationError({
2024 'location': _('Cannot assign stock to structural location')
2025 })
2027 data['location_id'] = location.pk
2029 # Set the parent ID correctly
2030 data['parent'] = self
2031 data['tree_id'] = self.tree_id
2033 # Generate a new serial number for each item
2034 items = StockItem._create_serial_numbers(serials, **data)
2036 # Create a new tracking entry for each item
2037 history_items = []
2039 for item in items:
2040 # Construct tracking entries for the new StockItem
2041 if entry := item.add_tracking_entry(
2042 StockHistoryCode.SPLIT_FROM_PARENT,
2043 user,
2044 quantity=1,
2045 notes=notes,
2046 location=location,
2047 commit=False,
2048 ):
2049 history_items.append(entry)
2051 if entry := item.add_tracking_entry(
2052 StockHistoryCode.ASSIGNED_SERIAL,
2053 user,
2054 notes=notes,
2055 deltas={'serial': item.serial},
2056 location=location,
2057 commit=False,
2058 ):
2059 history_items.append(entry)
2061 # Copy any test results from this item to the new one
2062 item.copyTestResultsFrom(self)
2064 StockItemTracking.objects.bulk_create(history_items, batch_size=250)
2066 # Remove the equivalent number of items
2067 self.take_stock(
2068 quantity, user, code=StockHistoryCode.STOCK_SERIALIZED, notes=notes
2069 )
2071 return items
2073 @transaction.atomic
2074 def copyHistoryFrom(self, other):
2075 """Copy stock history from another StockItem."""
2076 for item in other.tracking_info.all():
2077 item.item = self
2078 item.pk = None
2079 item.save()
2081 @transaction.atomic
2082 def copyTestResultsFrom(self, other: StockItem, filters: dict | None = None):
2083 """Copy all test results from another StockItem."""
2084 # Set default - see B006
2086 results = other.test_results.all()
2088 if filters:
2089 results = results.filter(**filters)
2091 results_to_create = []
2093 for result in list(results):
2094 # Create a copy of the test result by nulling-out the pk
2095 result.pk = None
2096 result.stock_item = self
2097 results_to_create.append(result)
2099 StockItemTestResult.objects.bulk_create(results_to_create, batch_size=250)
2101 def add_test_result(self, create_template=True, **kwargs):
2102 """Helper function to add a new StockItemTestResult.
2104 The main purpose of this function is to allow lookup of the template,
2105 based on the provided test name.
2107 If no template is found, a new one is created (if create_template=True).
2109 Args:
2110 create_template: If True, create a new template if it does not exist
2112 kwargs:
2113 template: The ID of the associated PartTestTemplate
2114 test_name: The name of the test (if the template is not provided)
2115 result: The result of the test
2116 value: The value of the test
2117 user: The user who performed the test
2118 notes: Any notes associated with the test
2119 """
2120 template = kwargs.get('template')
2121 test_name = kwargs.pop('test_name', None)
2123 test_key = InvenTree.helpers.generateTestKey(test_name)
2125 if template is None and test_name is not None:
2126 # Attempt to find a matching template
2128 ancestors = self.part.get_ancestors(include_self=True)
2130 template = PartModels.PartTestTemplate.objects.filter(
2131 part__tree_id=self.part.tree_id, part__in=ancestors, key=test_key
2132 ).first()
2134 if template is None:
2135 if create_template:
2136 template = PartModels.PartTestTemplate.objects.create(
2137 part=self.part, test_name=test_name
2138 )
2139 else:
2140 raise ValidationError({
2141 'template': _('Test template does not exist')
2142 })
2144 kwargs['template'] = template
2145 kwargs['stock_item'] = self
2147 return StockItemTestResult.objects.create(**kwargs)
2149 def can_merge(self, other=None, raise_error=False, **kwargs):
2150 """Check if this stock item can be merged into another stock item."""
2151 allow_mismatched_suppliers = kwargs.get('allow_mismatched_suppliers', False)
2153 allow_mismatched_status = kwargs.get('allow_mismatched_status', False)
2155 try:
2156 # Generic checks (do not rely on the 'other' part)
2157 if self.sales_order:
2158 raise ValidationError(
2159 _('Stock item has been assigned to a sales order')
2160 )
2162 if self.belongs_to:
2163 raise ValidationError(_('Stock item is installed in another item'))
2165 if self.installed_item_count() > 0:
2166 raise ValidationError(_('Stock item contains other items'))
2168 if self.customer:
2169 raise ValidationError(_('Stock item has been assigned to a customer'))
2171 if self.is_building:
2172 raise ValidationError(_('Stock item is currently in production'))
2174 if self.serialized:
2175 raise ValidationError(_('Serialized stock cannot be merged'))
2177 if other:
2178 # Specific checks (rely on the 'other' part)
2180 # Prevent stock item being merged with itself
2181 if self == other:
2182 raise ValidationError(_('Duplicate stock items'))
2184 # Base part must match
2185 if self.part != other.part:
2186 raise ValidationError(_('Stock items must refer to the same part'))
2188 # Check if supplier part references match
2189 if (
2190 self.supplier_part != other.supplier_part
2191 and not allow_mismatched_suppliers
2192 ):
2193 raise ValidationError(
2194 _('Stock items must refer to the same supplier part')
2195 )
2197 # Check if stock status codes match
2198 if self.status != other.status and not allow_mismatched_status:
2199 raise ValidationError(_('Stock status codes must match'))
2201 except ValidationError as e:
2202 if raise_error:
2203 raise e
2204 else:
2205 return False
2207 return True
2209 @transaction.atomic
2210 def merge_stock_items(self, other_items, raise_error=False, **kwargs):
2211 """Merge another stock item into this one; the two become one!
2213 *This* stock item subsumes the other, which is essentially deleted:
2215 - The quantity of this StockItem is increased
2216 - Tracking history for the *other* item is deleted
2217 - Any allocations (build order, sales order) are moved to this StockItem
2218 """
2219 if isinstance(other_items, StockItem):
2220 other_items = [other_items]
2222 if len(other_items) == 0:
2223 return
2225 # Keep track of the tree IDs that are being merged
2226 tree_ids = {self.tree_id}
2228 user = kwargs.get('user')
2229 location = kwargs.get('location', self.location)
2230 notes = kwargs.get('notes')
2232 parent_id = self.parent.pk if self.parent else None
2234 # Keep track of pricing data for the merged data
2235 pricing_data = []
2237 if self.purchase_price:
2238 pricing_data.append([self.purchase_price, self.quantity])
2240 for other in other_items:
2241 # If the stock item cannot be merged, return
2242 if not self.can_merge(other, raise_error=raise_error, **kwargs):
2243 logger.warning(
2244 'Stock item <%s> could not be merge into <%s>', other.pk, self.pk
2245 )
2246 return
2248 for other in other_items:
2249 tree_ids.add(other.tree_id)
2251 self.quantity += other.quantity
2253 if other.purchase_price:
2254 # Only add pricing data if it is available
2255 pricing_data.append([other.purchase_price, other.quantity])
2257 # Any "build order allocations" for the other item must be assigned to this one
2258 for allocation in other.allocations.all():
2259 allocation.stock_item = self
2260 allocation.save()
2262 # Any "sales order allocations" for the other item must be assigned to this one
2263 for allocation in other.sales_order_allocations.all():
2264 allocation.stock_item = self
2265 allocation.save()
2267 # Prevent atomicity issues when we are merging our own "parent" part in
2268 if parent_id and parent_id == other.pk:
2269 self.parent = None
2270 self.save()
2272 other.delete()
2274 self.add_tracking_entry(
2275 StockHistoryCode.MERGED_STOCK_ITEMS,
2276 user,
2277 quantity=self.quantity,
2278 notes=notes,
2279 deltas={
2280 'location': location.pk if location else None,
2281 'quantity': self.quantity,
2282 },
2283 )
2285 # Update the location of the item
2286 self.location = location
2288 # Update the unit price - calculate weighted average of available pricing data
2289 if len(pricing_data) > 0:
2290 unit_price, quantity = pricing_data[0]
2292 # Use the first currency as the base currency
2293 base_currency = unit_price.currency
2295 total_price = unit_price * quantity
2297 for price, qty in pricing_data[1:]:
2298 # Attempt to convert the price to the base currency
2299 try:
2300 price = convert_money(price, base_currency)
2301 total_price += price * qty
2302 quantity += qty
2303 except Exception:
2304 # Skip this entry, cannot convert to base currency
2305 continue
2307 if quantity > 0:
2308 self.purchase_price = total_price / quantity
2310 self.save()
2312 # Rebuild stock trees as required
2313 rebuild_result = True
2314 for tree_id in tree_ids:
2315 if not stock.tasks.rebuild_stock_item_tree(tree_id, rebuild_on_fail=False):
2316 rebuild_result = False
2318 if not rebuild_result:
2319 # If the rebuild failed, offload the task to a background worker
2320 logger.warning(
2321 'Failed to rebuild stock item tree during merge_stock_items operation, offloading task.'
2322 )
2323 InvenTree.tasks.offload_task(stock.tasks.rebuild_stock_items, group='stock')
2325 @transaction.atomic
2326 def splitStock(self, quantity, location=None, user=None, **kwargs):
2327 """Split this stock item into two items, in the same location.
2329 Stock tracking notes for this StockItem will be duplicated,
2330 and added to the new StockItem.
2332 Args:
2333 quantity: Number of stock items to remove from this entity, and pass to the next
2334 location: Where to move the new StockItem to
2335 user: User performing the action
2337 kwargs:
2338 notes: Optional notes for tracking
2339 batch: If provided, override the batch (default = existing batch)
2340 status: If provided, override the status (default = existing status)
2341 packaging: If provided, override the packaging (default = existing packaging)
2342 allow_production: If True, allow splitting of stock which is in production (default = False)
2344 Returns:
2345 The new StockItem object
2347 Raises:
2348 ValidationError: If the stock item cannot be split
2350 - The provided quantity will be subtracted from this item and given to the new one.
2351 - The new item will have a different StockItem ID, while this will remain the same.
2352 """
2353 # Run initial checks to test if the stock item can actually be "split"
2354 allow_production = kwargs.get('allow_production', False)
2356 # Cannot split a stock item which is in production
2357 if self.is_building and not allow_production:
2358 raise ValidationError(_('Stock item is currently in production'))
2360 notes = kwargs.get('notes', '')
2362 # Do not split a serialized part
2363 if self.serialized:
2364 return self
2366 try:
2367 quantity = Decimal(quantity)
2368 except (InvalidOperation, ValueError):
2369 return self
2371 # Doesn't make sense for a zero quantity
2372 if quantity <= 0:
2373 return self
2375 # Also doesn't make sense to split the full amount
2376 if quantity >= self.quantity:
2377 return self
2379 # Create a new StockItem object, duplicating relevant fields
2380 # Nullify the PK so a new record is created
2381 new_stock = StockItem.objects.get(pk=self.pk)
2382 new_stock.pk = None
2383 new_stock.quantity = quantity
2385 # Update the new stock item to ensure the tree structure is observed
2386 new_stock.parent = self
2387 new_stock.tree_id = None
2389 # Move to the new location if specified, otherwise use current location
2390 if location:
2391 new_stock.location = location
2392 else:
2393 new_stock.location = self.location
2395 deltas = {'stockitem': self.pk}
2397 transferorder = kwargs.pop('transferorder', None)
2398 if transferorder:
2399 deltas['transferorder'] = transferorder.pk
2401 # Optional fields which can be supplied in a 'move' call
2402 for field in StockItem.optional_transfer_fields():
2403 if field in kwargs:
2404 # handle specific case for status deltas
2405 if field == 'status':
2406 status = kwargs[field]
2407 if not new_stock.compare_status(status):
2408 old_custom_status = new_stock.get_custom_status()
2409 old_status_logical = new_stock.status
2410 new_stock.set_status(status)
2411 deltas['status'] = status # may be a custom value
2412 deltas['status_logical'] = (
2413 new_stock.status
2414 ) # always the logical value
2415 deltas['old_status'] = (
2416 old_custom_status
2417 if old_custom_status
2418 else old_status_logical
2419 )
2420 deltas['old_status_logical'] = old_status_logical
2421 else:
2422 setattr(new_stock, field, kwargs[field])
2423 deltas[field] = kwargs[field]
2425 new_stock.save(add_note=False)
2427 # Add a stock tracking entry for the newly created item
2428 new_stock.add_tracking_entry(
2429 StockHistoryCode.SPLIT_FROM_PARENT,
2430 user,
2431 quantity=quantity,
2432 notes=notes,
2433 location=location,
2434 deltas=deltas,
2435 )
2437 # Copy the test results of this part to the new one
2438 new_stock.copyTestResultsFrom(self)
2440 # Remove the specified quantity from THIS stock item
2441 self.take_stock(
2442 quantity,
2443 user,
2444 code=StockHistoryCode.SPLIT_CHILD_ITEM,
2445 notes=notes,
2446 location=location,
2447 stockitem=new_stock,
2448 )
2450 # Rebuild the tree for this parent item
2451 stock.tasks.rebuild_stock_item_tree(self.tree_id)
2453 # Attempt to reload the new item from the database
2454 try:
2455 new_stock.refresh_from_db()
2456 except Exception:
2457 pass
2459 trigger_event(StockEvents.ITEM_SPLIT, id=new_stock.id, parent=self.id)
2461 # Return a copy of the "new" stock item
2462 return new_stock
2464 @classmethod
2465 def optional_transfer_fields(cls):
2466 """Returns a list of optional fields for a stock transfer."""
2467 return ['batch', 'status', 'packaging']
2469 @transaction.atomic
2470 def move(self, location, notes, user, **kwargs):
2471 """Move part to a new location.
2473 If less than the available quantity is to be moved,
2474 a new StockItem is created, with the defined quantity,
2475 and that new StockItem is moved.
2476 The quantity is also subtracted from the existing StockItem.
2478 Args:
2479 location: Destination location (cannot be null)
2480 notes: User notes
2481 user: Who is performing the move
2483 kwargs:
2484 quantity: If provided, override the quantity (default = total stock quantity)
2485 batch: If provided, override the batch (default = existing batch)
2486 status: If provided, override the status (default = existing status)
2487 packaging: If provided, override the packaging (default = existing packaging)
2488 """
2489 current_location = self.location
2491 try:
2492 quantity = Decimal(kwargs.pop('quantity', self.quantity))
2493 except InvalidOperation:
2494 return False
2496 allow_out_of_stock_transfer = get_global_setting(
2497 'STOCK_ALLOW_OUT_OF_STOCK_TRANSFER', backup_value=False, cache=False
2498 )
2500 if not allow_out_of_stock_transfer and not self.is_in_stock(
2501 check_status=False, check_in_production=False
2502 ):
2503 raise ValidationError(_('StockItem cannot be moved as it is not in stock'))
2505 if quantity <= 0:
2506 return False
2508 if location is None:
2509 return False
2511 # Test for a partial movement
2512 if quantity < self.quantity:
2513 # We need to split the stock!
2515 kwargs['notes'] = notes
2517 # Split the existing StockItem in two
2518 self.splitStock(quantity, location, user, allow_production=True, **kwargs)
2520 return True
2522 # Moving into the same location triggers a different history code
2523 same_location = location == self.location
2525 self.location = location
2527 tracking_info = {'quantity': float(quantity)}
2529 tracking_code = StockHistoryCode.STOCK_MOVE
2531 if same_location:
2532 tracking_code = StockHistoryCode.STOCK_UPDATE
2533 else:
2534 tracking_info['location'] = location.pk
2536 status = kwargs.pop('status', None) or kwargs.pop('status_custom_key', None)
2538 if status and not self.compare_status(status):
2539 old_custom_status = self.get_custom_status()
2540 old_status_logical = self.status
2541 self.set_status(status)
2542 tracking_info['status'] = status # may be a custom value
2543 tracking_info['status_logical'] = self.status # always the logical value
2544 tracking_info['old_status'] = (
2545 old_custom_status if old_custom_status else old_status_logical
2546 )
2547 tracking_info['old_status_logical'] = old_status_logical
2549 transferorder = kwargs.pop('transferorder', None)
2550 if transferorder:
2551 tracking_info['transferorder'] = transferorder.pk
2553 # Optional fields which can be supplied in a 'move' call
2554 for field in StockItem.optional_transfer_fields():
2555 if field in kwargs:
2556 setattr(self, field, kwargs[field])
2557 tracking_info[field] = kwargs[field]
2559 self.add_tracking_entry(tracking_code, user, notes=notes, deltas=tracking_info)
2561 self.save(add_note=False)
2563 # Trigger event for the plugin system
2564 trigger_event(
2565 StockEvents.ITEM_MOVED,
2566 id=self.id,
2567 old_location=current_location.id if current_location else None,
2568 new_location=location.id if location else None,
2569 quantity=quantity,
2570 )
2572 return True
2574 @transaction.atomic
2575 def updateQuantity(self, quantity):
2576 """Update stock quantity for this item.
2578 If the quantity has reached zero, this StockItem will be deleted.
2580 Returns:
2581 - True if the quantity was saved
2582 - False if the StockItem was deleted
2583 """
2584 # Do not adjust quantity of a serialized part
2585 if self.serialized:
2586 return
2588 try:
2589 self.quantity = Decimal(quantity)
2590 except (InvalidOperation, ValueError):
2591 return
2593 quantity = max(quantity, 0)
2595 self.quantity = quantity
2597 if quantity == 0 and self.delete_on_deplete and self.can_delete():
2598 self.delete()
2600 return False
2602 self.save(add_note=False)
2604 trigger_event(
2605 StockEvents.ITEM_QUANTITY_UPDATED, id=self.id, quantity=float(self.quantity)
2606 )
2608 return True
2610 @transaction.atomic
2611 def stocktake(self, count, user, **kwargs):
2612 """Perform item stocktake.
2614 Arguments:
2615 count: The new quantity of the item
2616 user: The user performing the stocktake
2618 Keyword Arguments:
2619 notes: Optional notes for the stocktake
2620 status: Optionally adjust the stock status
2621 location: Optionally set the stock location
2622 """
2623 try:
2624 count = Decimal(count)
2625 except InvalidOperation:
2626 return False
2628 if count < 0:
2629 return False
2631 tracking_info = {}
2633 location = kwargs.pop('location', None)
2635 if location and location != self.location:
2636 old_location = self.location
2637 self.location = location
2638 tracking_info['location'] = location.pk
2639 tracking_info['old_location'] = old_location.pk if old_location else None
2641 status = kwargs.pop('status', None) or kwargs.pop('status_custom_key', None)
2643 if status and not self.compare_status(status):
2644 old_custom_status = self.get_custom_status()
2645 old_status_logical = self.status
2646 self.set_status(status)
2647 tracking_info['status'] = status # may be a custom value
2648 tracking_info['status_logical'] = self.status # always the logical value
2649 tracking_info['old_status'] = (
2650 old_custom_status if old_custom_status else old_status_logical
2651 )
2652 tracking_info['old_status_logical'] = old_status_logical
2654 if self.serialized or self.updateQuantity(count):
2655 tracking_info['quantity'] = 1 if self.serialized else float(count)
2657 self.stocktake_date = InvenTree.helpers.current_date()
2658 self.stocktake_user = user
2660 # Optional fields which can be supplied in a 'stocktake' call
2661 for field in StockItem.optional_transfer_fields():
2662 if field in kwargs:
2663 setattr(self, field, kwargs[field])
2664 tracking_info[field] = kwargs[field]
2666 self.save(add_note=False)
2668 self.add_tracking_entry(
2669 StockHistoryCode.STOCK_COUNT,
2670 user,
2671 notes=kwargs.get('notes', ''),
2672 deltas=tracking_info,
2673 )
2675 trigger_event(
2676 StockEvents.ITEM_COUNTED,
2677 'stockitem.counted',
2678 id=self.id,
2679 quantity=1 if self.serialized else float(self.quantity),
2680 )
2682 return True
2684 @transaction.atomic
2685 def add_stock(self, quantity, user, **kwargs):
2686 """Add a specified quantity of stock to this item.
2688 Arguments:
2689 quantity: The quantity to add
2690 user: The user performing the action
2692 Keyword Arguments:
2693 notes: Optional notes for the stock addition
2694 status: Optionally adjust the stock status
2695 """
2696 # Cannot add items to a serialized part
2697 if self.serialized:
2698 return False
2700 try:
2701 quantity = Decimal(quantity)
2702 except InvalidOperation:
2703 return False
2705 # Ignore amounts that do not make sense
2706 if quantity <= 0:
2707 return False
2709 tracking_info = {}
2711 status = kwargs.pop('status', None) or kwargs.pop('status_custom_key', None)
2713 if status and not self.compare_status(status):
2714 old_custom_status = self.get_custom_status()
2715 old_status_logical = self.status
2716 self.set_status(status)
2717 tracking_info['status'] = status # may be a custom value
2718 tracking_info['status_logical'] = self.status # always the logical value
2719 tracking_info['old_status'] = (
2720 old_custom_status if old_custom_status else old_status_logical
2721 )
2722 tracking_info['old_status_logical'] = old_status_logical
2724 if self.updateQuantity(self.quantity + quantity):
2725 tracking_info['added'] = float(quantity)
2726 tracking_info['quantity'] = float(self.quantity)
2728 # Optional fields which can be supplied in a 'stocktake' call
2729 for field in StockItem.optional_transfer_fields():
2730 if field in kwargs:
2731 setattr(self, field, kwargs[field])
2732 tracking_info[field] = kwargs[field]
2734 self.save(add_note=False)
2736 self.add_tracking_entry(
2737 StockHistoryCode.STOCK_ADD,
2738 user,
2739 notes=kwargs.get('notes', ''),
2740 deltas=tracking_info,
2741 )
2743 return True
2745 @transaction.atomic
2746 def take_stock(self, quantity, user, code=StockHistoryCode.STOCK_REMOVE, **kwargs):
2747 """Remove the specified quantity from this StockItem.
2749 Arguments:
2750 quantity: The quantity to remove
2751 user: The user performing the action
2753 Keyword Arguments:
2754 code: The stock history code to use
2755 notes: Optional notes for the stock removal
2756 status: Optionally adjust the stock status
2757 """
2758 # Cannot remove items from a serialized part
2759 if self.serialized:
2760 return False
2762 try:
2763 quantity = Decimal(quantity)
2764 except InvalidOperation:
2765 return False
2767 if quantity <= 0:
2768 return False
2770 deltas = {}
2772 status = kwargs.pop('status', None) or kwargs.pop('status_custom_key', None)
2774 if status and not self.compare_status(status):
2775 old_custom_status = self.get_custom_status()
2776 old_status_logical = self.status
2777 self.set_status(status)
2778 deltas['status'] = status # may be a custom value
2779 deltas['status_logical'] = self.status # always the logical value
2780 deltas['old_status'] = (
2781 old_custom_status if old_custom_status else old_status_logical
2782 )
2783 deltas['old_status_logical'] = old_status_logical
2785 if self.updateQuantity(self.quantity - quantity):
2786 deltas['removed'] = float(quantity)
2787 deltas['quantity'] = float(self.quantity)
2789 if location := kwargs.get('location'):
2790 deltas['location'] = location.pk
2792 if stockitem := kwargs.get('stockitem'):
2793 deltas['stockitem'] = stockitem.pk
2795 # Optional fields which can be supplied in a 'stocktake' call
2796 for field in StockItem.optional_transfer_fields():
2797 if field in kwargs:
2798 setattr(self, field, kwargs[field])
2799 deltas[field] = kwargs[field]
2801 transferorder = kwargs.pop('transferorder', None)
2802 if transferorder:
2803 deltas['transferorder'] = transferorder.pk
2805 self.save(add_note=False)
2807 self.add_tracking_entry(
2808 code, user, notes=kwargs.get('notes', ''), deltas=deltas
2809 )
2811 return True
2813 def __str__(self):
2814 """Human friendly name."""
2815 if self.part.trackable and self.serial:
2816 s = f'{self.part.full_name} #{self.serial}'
2817 else:
2818 s = f'{InvenTree.helpers.decimal2string(self.quantity)} x {self.part.full_name}'
2820 if self.location:
2821 s += f' @ {self.location.name}'
2823 if self.purchase_order:
2824 s += f' ({self.purchase_order})'
2826 return s
2828 @transaction.atomic
2829 def clear_test_results(self, **kwargs):
2830 """Remove all test results."""
2831 # All test results
2832 results = self.test_results.all()
2833 results.delete()
2835 def getTestResults(self, template=None, test=None, result=None, user=None):
2836 """Return all test results associated with this StockItem.
2838 Optionally can filter results by:
2839 - Test template ID
2840 - Test name
2841 - Test result
2842 - User
2843 """
2844 results = self.test_results
2846 if template:
2847 results = results.filter(template=template)
2849 if test:
2850 # Filter by test name
2851 test_key = InvenTree.helpers.generateTestKey(test)
2852 results = results.filter(template__key=test_key)
2854 if result is not None:
2855 # Filter by test status
2856 results = results.filter(result=result)
2858 if user:
2859 # Filter by user
2860 results = results.filter(user=user)
2862 return results
2864 def testResultMap(self, **kwargs):
2865 """Return a map of test-results using the test name as the key.
2867 Where multiple test results exist for a given name,
2868 the *most recent* test is used.
2870 This map is useful for rendering to a template (e.g. a test report),
2871 as all named tests are accessible.
2872 """
2873 # Do we wish to include test results from installed items?
2874 include_installed = kwargs.pop('include_installed', False)
2875 cascade = kwargs.pop('cascade', False)
2877 # Filter results by "date", so that newer results
2878 # will override older ones.
2879 results = self.getTestResults(**kwargs).order_by('date')
2881 result_map = {}
2883 for result in results:
2884 result_map[result.key] = result
2886 if include_installed:
2887 installed_items = self.get_installed_items(cascade=cascade)
2889 for item in installed_items:
2890 item_results = item.testResultMap()
2892 for key in item_results:
2893 # Results from sub items should not override master ones
2894 if key not in result_map:
2895 result_map[key] = item_results[key]
2897 return result_map
2899 def testResultList(self, **kwargs):
2900 """Return a list of test-result objects for this StockItem."""
2901 return list(self.testResultMap(**kwargs).values())
2903 def requiredTestStatus(self, required_tests=None):
2904 """Return the status of the tests required for this StockItem.
2906 Return:
2907 A dict containing the following items:
2908 - total: Number of required tests
2909 - passed: Number of tests that have passed
2910 - failed: Number of tests that have failed
2911 """
2912 # All the tests required by the part object
2914 if required_tests is None:
2915 required_tests = self.part.getRequiredTests()
2917 results = self.testResultMap()
2919 total = len(required_tests)
2920 passed = 0
2921 failed = 0
2923 for test in required_tests:
2924 key = InvenTree.helpers.generateTestKey(test.test_name)
2926 if key in results:
2927 result = results[key]
2929 if result.result:
2930 passed += 1
2931 else:
2932 failed += 1
2934 return {'total': total, 'passed': passed, 'failed': failed}
2936 @property
2937 def required_test_count(self):
2938 """Return the number of 'required tests' for this StockItem."""
2939 return self.part.getRequiredTests().count()
2941 def hasRequiredTests(self):
2942 """Return True if there are any 'required tests' associated with this StockItem."""
2943 return self.required_test_count > 0
2945 def passedAllRequiredTests(self, required_tests=None):
2946 """Returns True if this StockItem has passed all required tests."""
2947 status = self.requiredTestStatus(required_tests=required_tests)
2949 return status['passed'] >= status['total']
2952@receiver(post_delete, sender=StockItem, dispatch_uid='stock_item_post_delete_log')
2953def after_delete_stock_item(sender, instance: StockItem, **kwargs):
2954 """Function to be executed after a StockItem object is deleted."""
2955 from part import tasks as part_tasks
2957 if InvenTree.ready.isImportingData():
2958 return
2960 if InvenTree.ready.canAppAccessDatabase(allow_test=True):
2961 # Run this check in the background
2962 InvenTree.tasks.offload_task(
2963 part_tasks.notify_low_stock_if_required,
2964 instance.part.pk,
2965 group='notification',
2966 force_async=True,
2967 )
2969 if InvenTree.ready.canAppAccessDatabase(allow_test=settings.TESTING_PRICING):
2970 # Schedule an update on parent part pricing
2971 if instance.part:
2972 instance.part.schedule_pricing_update(create=False)
2975@receiver(post_save, sender=StockItem, dispatch_uid='stock_item_post_save_log')
2976def after_save_stock_item(sender, instance: StockItem, created, **kwargs):
2977 """Hook function to be executed after StockItem object is saved/updated."""
2978 from part import tasks as part_tasks
2980 if not InvenTree.ready.isImportingData():
2981 if InvenTree.ready.canAppAccessDatabase(allow_test=True):
2982 InvenTree.tasks.offload_task(
2983 part_tasks.notify_low_stock_if_required,
2984 instance.part.pk,
2985 group='notification',
2986 force_async=True,
2987 )
2989 if InvenTree.ready.canAppAccessDatabase(allow_test=settings.TESTING_PRICING):
2990 if instance.part:
2991 instance.part.schedule_pricing_update(create=True)
2994class StockItemTracking(InvenTree.models.InvenTreeModel):
2995 """Stock tracking entry - used for tracking history of a particular StockItem.
2997 Attributes:
2998 item: ForeignKey reference to a particular StockItem
2999 part: ForeignKey reference to the Part associated with this StockItem
3000 date: Date that this tracking info was created
3001 tracking_type: The type of tracking information
3002 notes: Associated notes (input by user)
3003 user: The user associated with this tracking info
3004 deltas: The changes associated with this history item
3006 Notes:
3007 If the underlying stock item is deleted, the "item" field will be set to null, but the tracking information will be retained.
3008 The tracking data will be removed if the associated part is deleted, as the tracking information is not relevant without the part context.
3009 """
3011 class Meta:
3012 """Meta data for the StockItemTracking class."""
3014 verbose_name = _('Stock Item Tracking')
3016 @staticmethod
3017 def get_api_url():
3018 """Return API url."""
3019 return reverse('api-stock-tracking-list')
3021 def save(self, *args, **kwargs):
3022 """Ensure that the 'part' link is always correct."""
3023 if self.item:
3024 self.part = self.item.part
3026 super().save(*args, **kwargs)
3028 def get_absolute_url(self):
3029 """Return url for instance."""
3030 return InvenTree.helpers.pui_url(f'/stock/item/{self.item.id}')
3032 def label(self):
3033 """Return label."""
3034 if self.tracking_type in StockHistoryCode.keys(): # noqa: SIM118
3035 return StockHistoryCode.label(self.tracking_type)
3037 return getattr(self, 'title', '')
3039 tracking_type = models.IntegerField(default=StockHistoryCode.LEGACY)
3041 item = models.ForeignKey(
3042 StockItem,
3043 on_delete=models.SET_NULL,
3044 null=True,
3045 blank=False,
3046 related_name='tracking_info',
3047 )
3049 part = models.ForeignKey(
3050 'part.part',
3051 on_delete=models.CASCADE,
3052 related_name='stock_tracking_info',
3053 null=True,
3054 blank=True,
3055 )
3057 date = models.DateTimeField(auto_now_add=True, editable=False)
3059 notes = models.CharField(
3060 blank=True,
3061 null=True,
3062 max_length=512,
3063 verbose_name=_('Notes'),
3064 help_text=_('Entry notes'),
3065 )
3067 user = models.ForeignKey(User, on_delete=models.SET_NULL, blank=True, null=True)
3069 deltas = models.JSONField(null=True, blank=True)
3072def rename_stock_item_test_result_attachment(instance, filename):
3073 """Rename test result."""
3074 return os.path.join(
3075 'stock_files', str(instance.stock_item.pk), os.path.basename(filename)
3076 )
3079class StockItemTestResult(InvenTree.models.InvenTreeMetadataModel):
3080 """A StockItemTestResult records results of custom tests against individual StockItem objects.
3082 This is useful for tracking unit acceptance tests, and particularly useful when integrated
3083 with automated testing setups.
3085 Multiple results can be recorded against any given test, allowing tests to be run many times.
3087 Attributes:
3088 stock_item: Link to StockItem
3089 template: Link to TestTemplate
3090 result: Test result value (pass / fail / etc)
3091 value: Recorded test output value (optional)
3092 attachment: Link to StockItem attachment (optional)
3093 notes: Extra user notes related to the test (optional)
3094 test_station: the name of the test station where the test was performed
3095 started_datetime: Date when the test was started
3096 finished_datetime: Date when the test was finished
3097 user: User who uploaded the test result
3098 date: Date the test result was recorded
3099 """
3101 class Meta:
3102 """Meta data for the StockItemTestResult class."""
3104 verbose_name = _('Stock Item Test Result')
3106 def __str__(self):
3107 """Return string representation."""
3108 return f'{self.test_name} - {self.result}'
3110 @staticmethod
3111 def get_api_url():
3112 """Return API url."""
3113 return reverse('api-stock-test-result-list')
3115 def save(self, *args, **kwargs):
3116 """Validate result is unique before saving."""
3117 super().clean()
3118 super().validate_unique()
3119 super().save(*args, **kwargs)
3121 def clean(self):
3122 """Make sure all values - including for templates - are provided."""
3123 super().clean()
3125 # If this test result corresponds to a template, check the requirements of the template
3126 try:
3127 template = self.template
3128 except PartModels.PartTestTemplate.DoesNotExist:
3129 template = None
3131 if not template:
3132 raise ValidationError({'template': _('Test template does not exist')})
3134 if template.requires_value and not self.value:
3135 raise ValidationError({'value': _('Value must be provided for this test')})
3137 if template.requires_attachment and not self.attachment:
3138 raise ValidationError({
3139 'attachment': _('Attachment must be uploaded for this test')
3140 })
3142 if choices := template.get_choices():
3143 if self.value not in choices:
3144 raise ValidationError({'value': _('Invalid value for this test')})
3146 @property
3147 def key(self):
3148 """Return key for test."""
3149 return InvenTree.helpers.generateTestKey(self.test_name)
3151 stock_item = models.ForeignKey(
3152 StockItem, on_delete=models.CASCADE, related_name='test_results'
3153 )
3155 @property
3156 def test_name(self):
3157 """Return the test name of the associated test template."""
3158 return self.template.test_name
3160 template = models.ForeignKey(
3161 'part.parttesttemplate',
3162 on_delete=models.CASCADE,
3163 blank=False,
3164 related_name='test_results',
3165 )
3167 result = models.BooleanField(
3168 default=False, verbose_name=_('Result'), help_text=_('Test result')
3169 )
3171 value = models.CharField(
3172 blank=True,
3173 max_length=500,
3174 verbose_name=_('Value'),
3175 help_text=_('Test output value'),
3176 )
3178 attachment = models.FileField(
3179 null=True,
3180 blank=True,
3181 upload_to=rename_stock_item_test_result_attachment,
3182 verbose_name=_('Attachment'),
3183 help_text=_('Test result attachment'),
3184 )
3186 notes = models.CharField(
3187 blank=True, max_length=500, verbose_name=_('Notes'), help_text=_('Test notes')
3188 )
3190 user = models.ForeignKey(User, on_delete=models.SET_NULL, blank=True, null=True)
3192 test_station = models.CharField(
3193 blank=True,
3194 max_length=500,
3195 verbose_name=_('Test station'),
3196 help_text=_('The identifier of the test station where the test was performed'),
3197 )
3199 started_datetime = models.DateTimeField(
3200 blank=True,
3201 null=True,
3202 verbose_name=_('Started'),
3203 help_text=_('The timestamp of the test start'),
3204 )
3206 finished_datetime = models.DateTimeField(
3207 blank=True,
3208 null=True,
3209 verbose_name=_('Finished'),
3210 help_text=_('The timestamp of the test finish'),
3211 )
3213 date = models.DateTimeField(
3214 default=InvenTree.helpers.current_time, verbose_name=_('Date')
3215 )