Coverage for src/backend/InvenTree/part/models.py: 31%
1690 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 17:47 +0000
1"""Part database model definitions."""
3from __future__ import annotations
5import hashlib
6import inspect
7import math
8import os
9import re
10from datetime import timedelta
11from decimal import ROUND_HALF_UP, Decimal, InvalidOperation
12from typing import TypedDict, cast
14from django.conf import settings
15from django.contrib.auth.models import User
16from django.contrib.contenttypes.models import ContentType
17from django.core.exceptions import ValidationError
18from django.core.validators import MaxValueValidator, MinValueValidator
19from django.db import models, transaction
20from django.db.models import F, Q, QuerySet, Sum, UniqueConstraint
21from django.db.models.functions import Coalesce
22from django.db.models.signals import post_delete, post_save
23from django.db.utils import IntegrityError
24from django.dispatch import receiver
25from django.urls import reverse
26from django.utils.translation import gettext_lazy as _
28import structlog
29from django_cleanup import cleanup
30from djmoney.contrib.exchange.exceptions import MissingRate
31from djmoney.contrib.exchange.models import convert_money
32from djmoney.money import Money
33from mptt.managers import TreeManager
34from mptt.models import TreeForeignKey
36import common.currency
37import common.models
38import InvenTree.conversion
39import InvenTree.fields
40import InvenTree.helpers
41import InvenTree.models
42import InvenTree.ready
43import InvenTree.tasks
44import part.helpers as part_helpers
45import part.settings as part_settings
46import report.mixins
47import users.models
48from build import models as BuildModels
49from build.status_codes import BuildStatusGroups
50from common.currency import currency_code_default
51from common.icons import validate_icon
52from common.settings import get_global_setting
53from InvenTree import helpers, validators
54from InvenTree.exceptions import log_error
55from InvenTree.fields import InvenTreeURLField
56from InvenTree.helpers import decimal2money, decimal2string, normalize
57from order import models as OrderModels
58from order.status_codes import (
59 PurchaseOrderStatus,
60 PurchaseOrderStatusGroups,
61 SalesOrderStatusGroups,
62 TransferOrderStatusGroups,
63)
64from stock import models as StockModels
66logger = structlog.get_logger('inventree')
69class PartCategory(
70 InvenTree.models.PluginValidationMixin,
71 InvenTree.models.InvenTreeParameterMixin,
72 InvenTree.models.MetadataMixin,
73 InvenTree.models.PathStringMixin,
74 InvenTree.models.InvenTreeTree,
75):
76 """PartCategory provides hierarchical organization of Part objects.
78 Attributes:
79 name: Name of this category
80 parent: Parent category
81 default_location: Default storage location for parts in this category or child categories
82 default_keywords: Default keywords for parts created in this category
83 """
85 ITEM_PARENT_KEY = 'category'
86 EXTRA_PATH_FIELDS = ['icon']
87 IMPORT_ID_FIELDS = ['pathstring', 'name']
89 class Meta:
90 """Metaclass defines extra model properties."""
92 verbose_name = _('Part Category')
93 verbose_name_plural = _('Part Categories')
95 def delete(self, *args, **kwargs):
96 """Custom model deletion routine, which updates any child categories or parts.
98 This must be handled within a transaction.atomic(), otherwise the tree structure is damaged
99 """
100 super().delete(
101 delete_children=kwargs.get('delete_child_categories', False),
102 delete_items=kwargs.get('delete_parts', False),
103 )
105 default_location = TreeForeignKey(
106 'stock.StockLocation',
107 related_name='default_categories',
108 null=True,
109 blank=True,
110 on_delete=models.SET_NULL,
111 verbose_name=_('Default Location'),
112 help_text=_('Default location for parts in this category'),
113 )
115 structural = models.BooleanField(
116 default=False,
117 verbose_name=_('Structural'),
118 help_text=_(
119 'Parts may not be directly assigned to a structural category, '
120 'but may be assigned to child categories.'
121 ),
122 )
124 default_keywords = models.CharField(
125 null=True,
126 blank=True,
127 max_length=250,
128 verbose_name=_('Default keywords'),
129 help_text=_('Default keywords for parts in this category'),
130 )
132 _icon = models.CharField(
133 blank=True,
134 null=True,
135 max_length=100,
136 verbose_name=_('Icon'),
137 help_text=_('Icon (optional)'),
138 validators=[validate_icon],
139 db_column='icon',
140 )
142 @property
143 def icon(self):
144 """Return the icon associated with this PartCategory or the default icon."""
145 if self._icon: 145 ↛ 146line 145 didn't jump to line 146 because the condition on line 145 was never true
146 return self._icon
148 if default_icon := get_global_setting('PART_CATEGORY_DEFAULT_ICON', cache=True): 148 ↛ 149line 148 didn't jump to line 149 because the condition on line 148 was never true
149 return default_icon
151 return ''
153 @icon.setter
154 def icon(self, value):
155 """Setter for icon field."""
156 default_icon = get_global_setting('PART_CATEGORY_DEFAULT_ICON', cache=True)
158 # if icon is not defined previously and new value is default icon, do not save it
159 if not self._icon and value == default_icon: 159 ↛ 162line 159 didn't jump to line 162 because the condition on line 159 was always true
160 return
162 self._icon = value
164 @staticmethod
165 def get_api_url():
166 """Return the API url associated with the PartCategory model."""
167 return reverse('api-part-category-list')
169 def get_absolute_url(self):
170 """Return the web URL associated with the detail view for this PartCategory instance."""
171 return helpers.pui_url(f'/part/category/{self.id}')
173 def clean(self):
174 """Custom clean action for the PartCategory model.
176 Ensure that the structural parameter cannot get set if products already assigned to the category
177 """
178 if self.pk and self.structural and self.partcount(False, False) > 0: 178 ↛ 179line 178 didn't jump to line 179 because the condition on line 178 was never true
179 raise ValidationError(
180 _(
181 'You cannot make this part category structural because some parts '
182 'are already assigned to it!'
183 )
184 )
185 super().clean()
187 def get_parts(self, cascade=True) -> set[Part]:
188 """Return a queryset for all parts under this category.
190 Args:
191 cascade (bool, optional): If True, also look under subcategories. Defaults to True.
193 Returns:
194 set[Part]: All matching parts
195 """
196 if cascade:
197 """Select any parts which exist in this category or any child categories."""
198 queryset = Part.objects.filter(
199 category__in=self.getUniqueChildren(include_self=True)
200 )
201 else:
202 queryset = Part.objects.filter(category=self.pk)
204 return queryset
206 @property
207 def item_count(self):
208 """Return the number of parts contained in this PartCategory."""
209 return self.partcount()
211 def get_items(self, cascade=False):
212 """Return a queryset containing the parts which exist in this category."""
213 return self.get_parts(cascade=cascade)
215 def partcount(self, cascade=True, active=False):
216 """Return the total part count under this category (including children of child categories)."""
217 query = self.get_parts(cascade=cascade)
219 if active:
220 query = query.filter(active=True)
222 return query.count()
224 def prefetch_parts_parameters(self, cascade=True):
225 """Prefectch parts parameters."""
226 return (
227 self
228 .get_parts(cascade=cascade)
229 .prefetch_related('parameters_list', 'parameters_list__template')
230 .all()
231 )
233 def get_unique_parameters(self, cascade=True, prefetch=None):
234 """Get all unique parameter names for all parts from this category."""
235 unique_parameters_names = []
237 parts = prefetch or self.prefetch_parts_parameters(cascade=cascade)
239 for part in parts:
240 for parameter in part.parameters_list.all():
241 parameter_name = parameter.template.name
242 if parameter_name not in unique_parameters_names:
243 unique_parameters_names.append(parameter_name)
245 return sorted(unique_parameters_names)
247 def get_parts_parameters(self, cascade=True, prefetch=None):
248 """Get all parameter names and values for all parts from this category."""
249 category_parameters = []
251 parts = prefetch or self.prefetch_parts_parameters(cascade=cascade)
253 for part in parts:
254 part_parameters = {
255 'pk': part.pk,
256 'name': part.name,
257 'description': part.description,
258 }
259 # Add IPN only if it exists
260 if part.IPN:
261 part_parameters['IPN'] = part.IPN
263 for parameter in part.parameters_list.all():
264 parameter_name = parameter.template.name
265 parameter_value = parameter.data
266 part_parameters[parameter_name] = parameter_value
268 category_parameters.append(part_parameters)
270 return category_parameters
272 @classmethod
273 def get_parent_categories(cls):
274 """Return tuple list of parent (root) categories."""
275 # Get root nodes
276 root_categories = cls.objects.filter(level=0)
278 parent_categories = []
279 for category in root_categories:
280 parent_categories.append((category.id, category.name))
282 return parent_categories
284 def get_parameter_templates(self):
285 """Return parameter templates associated to category."""
286 prefetch = PartCategoryParameterTemplate.objects.prefetch_related(
287 'category', 'parameter'
288 )
290 return prefetch.filter(category=self.id)
292 def get_subscribers(self, include_parents: bool = True) -> list[User]:
293 """Return a list of users who subscribe to this PartCategory.
295 Arguments:
296 include_parents (bool): If True, include users who subscribe to parent categories.
298 Returns:
299 list[User]: List of users who subscribe to this category.
300 """
301 subscribers = set()
303 if include_parents:
304 cats = self.get_ancestors(include_self=True)
305 queryset = PartCategoryStar.objects.filter(category__in=cats)
306 else:
307 queryset = PartCategoryStar.objects.filter(category=self)
309 for result in queryset:
310 subscribers.add(result.user)
312 return list(subscribers)
314 def is_starred_by(self, user, **kwargs):
315 """Returns True if the specified user subscribes to this category."""
316 return user in self.get_subscribers(**kwargs)
318 def set_starred(self, user, status: bool, **kwargs) -> None:
319 """Set the "subscription" status of this PartCategory against the specified user."""
320 if not user:
321 return
323 if self.is_starred_by(user, **kwargs) == status:
324 return
326 if status:
327 PartCategoryStar.objects.create(category=self, user=user)
328 else:
329 # Note that this won't actually stop the user being subscribed,
330 # if the user is subscribed to a parent category
331 PartCategoryStar.objects.filter(category=self, user=user).delete()
334def rename_part_image(instance, filename):
335 """Function for renaming a part image file.
337 Args:
338 instance: Instance of a Part object
339 filename: Name of original uploaded file
341 Returns:
342 Cleaned filename in format part_<n>_img
343 """
344 base = part_helpers.PART_IMAGE_DIR
345 fname = os.path.basename(filename)
347 return os.path.join(base, fname)
350class PartCategoryParameterTemplate(InvenTree.models.InvenTreeMetadataModel):
351 """A PartCategoryParameterTemplate creates a unique relationship between a PartCategory and a ParameterTemplate.
353 Multiple ParameterTemplate instances can be associated to a PartCategory to drive a default list of parameter templates attached to a Part instance upon creation.
355 Attributes:
356 category: Reference to a single PartCategory object
357 template: Reference to a single ParameterTemplate object
358 default_value: The default value for the parameter in the context of the selected category
359 """
361 @staticmethod
362 def get_api_url():
363 """Return the API endpoint URL associated with the PartCategoryParameterTemplate model."""
364 return reverse('api-part-category-parameter-list')
366 class Meta:
367 """Metaclass providing extra model definition."""
369 verbose_name = _('Part Category Parameter Template')
371 constraints = [
372 UniqueConstraint(
373 fields=['category', 'template'], name='unique_category_parameter_pair'
374 )
375 ]
377 def __str__(self):
378 """String representation of a PartCategoryParameterTemplate (admin interface)."""
379 if self.default_value:
380 return f'{self.category.name} | {self.template.name} | {self.default_value}'
381 return f'{self.category.name} | {self.template.name}'
383 def clean(self):
384 """Validate this PartCategoryParameterTemplate instance.
386 Checks the provided 'default_value', and (if not blank), ensure it is valid.
387 """
388 super().clean()
390 self.default_value = (
391 '' if self.default_value is None else str(self.default_value.strip())
392 )
394 if (
395 self.default_value
396 and get_global_setting(
397 'PARAMETER_ENFORCE_UNITS', True, cache=False, create=False
398 )
399 and self.template.units
400 ):
401 try:
402 InvenTree.conversion.convert_physical_value(
403 self.default_value, self.template.units
404 )
405 except ValidationError as e:
406 raise ValidationError({'default_value': e.message})
408 category = models.ForeignKey(
409 PartCategory,
410 on_delete=models.CASCADE,
411 related_name='parameter_templates',
412 verbose_name=_('Category'),
413 help_text=_('Part Category'),
414 )
416 template = models.ForeignKey(
417 common.models.ParameterTemplate,
418 on_delete=models.CASCADE,
419 related_name='part_categories',
420 )
422 default_value = models.CharField(
423 max_length=500,
424 blank=True,
425 verbose_name=_('Default Value'),
426 help_text=_('Default Parameter Value'),
427 )
430class PartReportContext(report.mixins.BaseReportContext, TypedDict):
431 """Report context for the Part model.
433 Attributes:
434 bom_items: Query set of all BomItem objects associated with the Part
435 category: The PartCategory object associated with the Part
436 description: The description field of the Part
437 IPN: The IPN (internal part number) of the Part
438 name: The name of the Part
439 parameters: Dict object containing the parameters associated with the Part
440 part: The Part object itself
441 qr_data: Formatted QR code data for the Part
442 qr_url: Generated URL for embedding in a QR code
443 revision: The revision of the Part
444 test_template_list: List of test templates associated with the Part
445 test_templates: Dict object of test templates associated with the Part
446 """
448 bom_items: report.mixins.QuerySet[BomItem]
449 category: PartCategory | None
450 description: str
451 IPN: str | None
452 name: str
453 parameters: dict[str, str]
454 part: Part
455 qr_data: str
456 qr_url: str
457 revision: str | None
458 test_template_list: report.mixins.QuerySet[PartTestTemplate]
459 test_templates: dict[str, PartTestTemplate]
462@cleanup.ignore
463class Part(
464 InvenTree.models.PluginValidationMixin,
465 InvenTree.models.InvenTreeParameterMixin,
466 InvenTree.models.InvenTreeAttachmentMixin,
467 InvenTree.models.InvenTreeBarcodeMixin,
468 InvenTree.models.InvenTreeTagsMixin,
469 InvenTree.models.InvenTreeNotesMixin,
470 report.mixins.InvenTreeReportMixin,
471 InvenTree.models.InvenTreeImageMixin,
472 InvenTree.models.MetadataMixin,
473 InvenTree.models.InvenTreeTree,
474):
475 """The Part object represents an abstract part, the 'concept' of an actual entity.
477 An actual physical instance of a Part is a StockItem which is treated separately.
479 Parts can be used to create other parts (as part of a Bill of Materials or BOM).
481 Attributes:
482 name: Brief name for this part
483 variant: Optional variant number for this part - Must be unique for the part name
484 category: The PartCategory to which this part belongs
485 description: Longer form description of the part
486 keywords: Optional keywords for improving part search results
487 IPN: Internal part number (optional)
488 revision: Part revision
489 is_template: If True, this part is a 'template' part
490 link: Link to an external page with more information about this part (e.g. internal Wiki)
491 image: Image of this part
492 default_location: Where the item is normally stored (may be null)
493 default_expiry: The default expiry duration for any StockItem instances of this part
494 minimum_stock: Minimum preferred quantity to keep in stock
495 maximum_stock: Maximum preferred quantity to keep in stock
496 units: Units of measure for this part (default='pcs')
497 salable: Can this part be sold to customers?
498 assembly: Can this part be build from other parts?
499 component: Can this part be used to make other parts?
500 purchaseable: Can this part be purchased from suppliers?
501 trackable: Trackable parts can have unique serial numbers assigned, etc, etc
502 testable: Testable parts can have test results recorded against their stock items
503 active: Is this part active? Parts are deactivated instead of being deleted
504 locked: This part is locked and cannot be edited
505 virtual: Is this part "virtual"? e.g. a software product or similar
506 notes: Additional notes field for this part
507 creation_date: Date that this part was added to the database
508 creation_user: User who added this part to the database
509 responsible_owner: Owner (either user or group) which is responsible for this part (optional)
511 BOM (Bill of Materials) related attributes:
512 bom_checksum: Checksum for the BOM of this part
513 bom_validated: Boolean field indicating if the BOM is valid (checksum matches)
514 bom_checked_by: User who last checked the BOM for this part
515 bom_checked_date: Date when the BOM was last checked
516 """
518 NODE_PARENT_KEY = 'variant_of'
519 IMAGE_RENAME = rename_part_image
520 IMPORT_ID_FIELDS = ['IPN', 'name']
522 objects = TreeManager()
524 class Meta:
525 """Metaclass defines extra model properties."""
527 verbose_name = _('Part')
528 verbose_name_plural = _('Parts')
529 ordering = ['name']
530 constraints = [
531 UniqueConstraint(fields=['name', 'IPN', 'revision'], name='unique_part')
532 ]
534 class MPTTMeta:
535 """MPTT Metaclass options."""
537 # For legacy reasons the 'variant_of' field is used to indicate the MPTT parent
538 parent_attr = 'variant_of'
540 @staticmethod
541 def get_api_url():
542 """Return the list API endpoint URL associated with the Part model."""
543 return reverse('api-part-list')
545 def api_instance_filters(self):
546 """Return API query filters for limiting field results against this instance."""
547 return {'variant_of': {'exclude_tree': self.pk}}
549 @classmethod
550 def barcode_model_type_code(cls):
551 """Return the associated barcode model type code for this model."""
552 return 'PA'
554 def report_context(self) -> PartReportContext:
555 """Return custom report context information."""
556 return {
557 'bom_items': cast(report.mixins.QuerySet['BomItem'], self.get_bom_items()),
558 'category': self.category,
559 'description': self.description,
560 'IPN': self.IPN,
561 'name': self.name,
562 'parameters': self.parameters_map(),
563 'part': self,
564 'qr_data': self.barcode,
565 'qr_url': self.get_absolute_url(),
566 'revision': self.revision,
567 'test_template_list': self.getTestTemplates(),
568 'test_templates': self.getTestTemplateMap(),
569 }
571 def check_parameter_delete(self, parameter):
572 """Custom delete check for Parameter instances associated with this Part."""
573 if self.locked and get_global_setting('PART_ENABLE_LOCKING'):
574 raise ValidationError(_('Cannot delete parameters of a locked part'))
576 def check_parameter_save(self, parameter):
577 """Custom save check for Parameter instances associated with this Part."""
578 if self.locked and get_global_setting('PART_ENABLE_LOCKING'):
579 raise ValidationError(_('Cannot modify parameters of a locked part'))
581 def delete(self, **kwargs):
582 """Custom delete method for the Part model.
584 Prevents deletion of a Part if any of the following conditions are met:
586 - The part is still active
587 - The part is used in a BOM for a different part.
588 """
589 if self.locked and get_global_setting('PART_ENABLE_LOCKING'): 589 ↛ 590line 589 didn't jump to line 590 because the condition on line 589 was never true
590 raise ValidationError(_('Cannot delete this part as it is locked'))
592 if self.active: 592 ↛ 595line 592 didn't jump to line 595 because the condition on line 592 was always true
593 raise ValidationError(_('Cannot delete this part as it is still active'))
595 if not get_global_setting('PART_ALLOW_DELETE_FROM_ASSEMBLY', cache=False):
596 if BomItem.objects.filter(sub_part=self).exists():
597 raise ValidationError(
598 _('Cannot delete this part as it is used in an assembly')
599 )
601 super().delete()
603 def save(self, *args, **kwargs):
604 """Overrides the save function for the Part model.
606 If the part image has been updated, then check if the "old" (previous) image is still used by another part.
607 If not, it is considered "orphaned" and will be deleted.
608 """
609 _new = False
610 if self.pk:
611 try:
612 previous = Part.objects.get(pk=self.pk)
614 # Image has been changed
615 if previous.image is not None and self.image != previous.image:
616 # Are there any (other) parts which reference the image?
617 n_refs = (
618 Part.objects
619 .filter(image=previous.image)
620 .exclude(pk=self.pk)
621 .count()
622 )
624 if n_refs == 0:
625 logger.info("Deleting unused image file '%s'", previous.image)
626 previous.image.delete(save=False)
627 except Part.DoesNotExist:
628 pass
629 else:
630 _new = True
632 self.full_clean()
634 super().save(*args, **kwargs)
636 if _new:
637 # Only run if the check was not run previously (due to not existing in the database)
638 self.ensure_trackable()
640 def __str__(self):
641 """Return a string representation of the Part (for use in the admin interface)."""
642 return f'{self.full_name} - {self.description}'
644 def get_parts_in_bom(self, **kwargs):
645 """Return a list of all parts in the BOM for this part.
647 Takes into account substitutes, variant parts, and inherited BOM items
648 """
649 parts = set()
651 for bom_item in self.get_bom_items(**kwargs):
652 for part in bom_item.get_valid_parts_for_allocation():
653 parts.add(part)
655 return parts
657 def check_if_part_in_bom(self, other_part, **kwargs):
658 """Check if the other_part is in the BOM for *this* part.
660 Note:
661 - Accounts for substitute parts
662 - Accounts for variant BOMs
663 """
664 return other_part in self.get_parts_in_bom(**kwargs)
666 def check_add_to_bom(self, parent, raise_error=False, recursive=True):
667 """Check if this Part can be added to the BOM of another part.
669 This will fail if:
671 a) The parent part is the same as this one
672 b) The parent part exists in the same variant tree as this one
673 c) The parent part is used in the BOM for *this* part
674 d) The parent part is used in the BOM for any child parts under this one
675 """
676 result = True
678 try:
679 if self.pk == parent.pk:
680 raise ValidationError({
681 'sub_part': _(
682 f"Part '{self}' cannot be used in BOM for '{parent}' (recursive)"
683 )
684 })
686 if self.tree_id == parent.tree_id:
687 raise ValidationError({
688 'sub_part': _(
689 f"Part '{self}' cannot be used in BOM for '{parent}' (recursive)"
690 )
691 })
693 bom_items = self.get_bom_items()
695 # Ensure that the parent part does not appear under any child BOM item!
696 for item in bom_items.all():
697 # Check for simple match
698 if item.sub_part == parent:
699 raise ValidationError({
700 'sub_part': _(
701 f"Part '{parent}' is used in BOM for '{self}' (recursive)"
702 )
703 })
705 # And recursively check too
706 if recursive:
707 result = result and item.sub_part.check_add_to_bom(
708 parent, recursive=True, raise_error=raise_error
709 )
711 except ValidationError as e:
712 if raise_error:
713 raise e
714 else:
715 return False
717 return result
719 def validate_name(self, raise_error=True):
720 """Validate the name field for this Part instance.
722 This function is exposed to any Validation plugins, and thus can be customized.
723 """
724 from plugin import PluginMixinEnum, registry
726 # Skip plugin validation checks during read-only management commands
727 if not InvenTree.ready.isReadOnlyCommand(): 727 ↛ exitline 727 didn't return from function 'validate_name' because the condition on line 727 was always true
728 for plugin in registry.with_mixin(PluginMixinEnum.VALIDATION): 728 ↛ 732line 728 didn't jump to line 732 because the loop on line 728 never started
729 # Run the name through each custom validator
730 # If the plugin returns 'True' we will skip any subsequent validation
732 try:
733 result = plugin.validate_part_name(self.name, self)
734 if result:
735 return
736 except ValidationError as exc:
737 if raise_error:
738 raise ValidationError({'name': exc.message})
739 except Exception:
740 log_error('validate_part_name', plugin=plugin.slug)
742 def validate_ipn(self, raise_error=True):
743 """Ensure that the IPN (internal part number) is valid for this Part".
745 - Validation is handled by custom plugins
746 - By default, no validation checks are performed
747 """
748 from plugin import PluginMixinEnum, registry
750 # Skip plugin validation checks during read-only management commands
751 if not InvenTree.ready.isReadOnlyCommand(): 751 ↛ 766line 751 didn't jump to line 766 because the condition on line 751 was always true
752 for plugin in registry.with_mixin(PluginMixinEnum.VALIDATION): 752 ↛ 753line 752 didn't jump to line 753 because the loop on line 752 never started
753 try:
754 result = plugin.validate_part_ipn(self.IPN, self)
756 if result:
757 # A "true" result force skips any subsequent checks
758 break
759 except ValidationError as exc:
760 if raise_error:
761 raise ValidationError({'IPN': exc.message})
762 except Exception:
763 log_error('validate_part_ipn', plugin=plugin.slug)
765 # If we get to here, none of the plugins have raised an error
766 pattern = get_global_setting('PART_IPN_REGEX', '', create=False).strip()
768 if pattern: 768 ↛ 769line 768 didn't jump to line 769 because the condition on line 768 was never true
769 match = re.search(pattern, self.IPN)
771 if match is None:
772 raise ValidationError(_(f'IPN must match regex pattern {pattern}'))
774 def validate_revision(self):
775 """Check the 'revision' and 'revision_of' fields."""
776 # Part cannot be a revision of itself
777 if self.revision_of: 777 ↛ 778line 777 didn't jump to line 778 because the condition on line 777 was never true
778 if self.revision_of == self:
779 raise ValidationError({
780 'revision_of': _('Part cannot be a revision of itself')
781 })
783 # If this part is a revision, it must have a revision code
784 if not self.revision:
785 raise ValidationError({
786 'revision': _(
787 'Revision code must be specified for a part marked as a revision'
788 )
789 })
791 if get_global_setting('PART_REVISION_ASSEMBLY_ONLY'):
792 if not self.assembly or not self.revision_of.assembly:
793 raise ValidationError({
794 'revision_of': _(
795 'Revisions are only allowed for assembly parts'
796 )
797 })
799 # Cannot have a revision of a "template" part
800 if self.revision_of.is_template:
801 raise ValidationError({
802 'revision_of': _('Cannot make a revision of a template part')
803 })
805 # parent part must point to the same template (via variant_of)
806 if self.variant_of != self.revision_of.variant_of:
807 raise ValidationError({
808 'revision_of': _('Parent part must point to the same template')
809 })
811 def validate_serial_number(
812 self,
813 serial: str,
814 stock_item=None,
815 check_duplicates=True,
816 raise_error=False,
817 **kwargs,
818 ):
819 """Validate a serial number against this Part instance.
821 Note: This function is exposed to any Validation plugins, and thus can be customized.
823 Any plugins which implement the 'validate_serial_number' method have three possible outcomes:
825 - Decide the serial is objectionable and raise a django.core.exceptions.ValidationError
826 - Decide the serial is acceptable, and return None to proceed to other tests
827 - Decide the serial is acceptable, and return True to skip any further tests
829 Arguments:
830 serial: The proposed serial number
831 stock_item: (optional) A StockItem instance which has this serial number assigned (e.g. testing for duplicates)
832 check_duplicates: If True, checks for duplicate serial numbers in the database.
833 raise_error: If False, and ValidationError(s) will be handled
835 Returns:
836 True if serial number is 'valid' else False
838 Raises:
839 ValidationError if serial number is invalid and raise_error = True
840 """
841 from plugin import PluginMixinEnum, registry
843 serial = str(serial).strip()
845 if not InvenTree.ready.isReadOnlyCommand():
846 # First, throw the serial number against each of the loaded validation plugins
847 for plugin in registry.with_mixin(PluginMixinEnum.VALIDATION):
848 # Run the serial number through each custom validator
849 # If the plugin returns 'True' we will skip any subsequent validation
851 try:
852 result = False
854 if hasattr(plugin, 'validate_serial_number'):
855 signature = inspect.signature(plugin.validate_serial_number)
857 if 'stock_item' in signature.parameters:
858 # 2024-08-21: New method signature accepts a 'stock_item' parameter
859 result = plugin.validate_serial_number(
860 serial, self, stock_item=stock_item
861 )
862 else:
863 # Old method signature - does not accept a 'stock_item' parameter
864 result = plugin.validate_serial_number(serial, self)
866 if result is True:
867 return True
868 except ValidationError as exc:
869 if raise_error:
870 # Re-throw the error
871 raise exc
872 else:
873 return False
874 except Exception:
875 log_error('validate_serial_number', plugin=plugin.slug)
877 """
878 If we are here, none of the loaded plugins (if any) threw an error or exited early
880 Now, we run the "default" serial number validation routine,
881 which checks that the serial number is not duplicated
882 """
884 if not check_duplicates:
885 return
887 # from part.models import Part
888 from stock.models import StockItem
890 if get_global_setting('SERIAL_NUMBER_GLOBALLY_UNIQUE', False):
891 # Serial number must be unique across *all* parts
892 parts = Part.objects.all()
893 else:
894 # Serial number must only be unique across this part "tree"
895 parts = Part.objects.filter(tree_id=self.tree_id)
897 stock = StockItem.objects.filter(part__in=parts, serial=serial)
899 if stock_item:
900 # Exclude existing StockItem from query
901 stock = stock.exclude(pk=stock_item.pk)
903 if stock.exists():
904 if raise_error:
905 raise ValidationError(
906 _('Stock item with this serial number already exists')
907 + ': '
908 + serial
909 )
910 else:
911 return False
912 else:
913 # This serial number is perfectly valid
914 return True
916 def find_conflicting_serial_numbers(self, serials: list) -> list:
917 """For a provided list of serials, return a list of those which are conflicting."""
918 # from part.models import Part
919 from stock.models import StockItem
921 conflicts = []
923 # First, check for raw conflicts based on efficient database queries
924 if get_global_setting('SERIAL_NUMBER_GLOBALLY_UNIQUE', False):
925 # Serial number must be unique across *all* parts
926 parts = Part.objects.all()
927 else:
928 # Serial number must only be unique across this part "tree"
929 parts = Part.objects.filter(tree_id=self.tree_id)
931 items = StockItem.objects.filter(part__in=parts, serial__in=serials)
932 items = items.order_by('serial_int', 'serial')
934 for item in items:
935 conflicts.append(item.serial)
937 for serial in serials:
938 if serial in conflicts:
939 # Already found a conflict, no need to check further
940 continue
942 try:
943 self.validate_serial_number(
944 serial, raise_error=True, check_duplicates=False
945 )
946 except ValidationError:
947 # Serial number is invalid (as determined by plugin)
948 conflicts.append(serial)
950 return conflicts
952 def get_latest_serial_number(self, allow_plugins=True):
953 """Find the 'latest' serial number for this Part.
955 Here we attempt to find the "highest" serial number which exists for this Part.
956 There are a number of edge cases where this method can fail,
957 but this is accepted to keep database performance at a reasonable level.
959 Note: Serial numbers must be unique across an entire Part "tree",
960 so we filter by the entire tree.
962 Returns:
963 The latest serial number specified for this part, or None
964 """
965 from plugin import PluginMixinEnum, registry
967 if allow_plugins and not InvenTree.ready.isReadOnlyCommand(): 967 ↛ 979line 967 didn't jump to line 979 because the condition on line 967 was always true
968 # Check with plugin system
969 # If any plugin returns a non-null result, that takes priority
970 for plugin in registry.with_mixin(PluginMixinEnum.VALIDATION): 970 ↛ 971line 970 didn't jump to line 971 because the loop on line 970 never started
971 try:
972 result = plugin.get_latest_serial_number(self)
973 if result is not None:
974 return str(result)
975 except Exception:
976 log_error('get_latest_serial_number', plugin=plugin.slug)
978 # No plugin returned a result, so we will run the default query
979 stock = (
980 StockModels.StockItem.objects.all().exclude(serial=None).exclude(serial='')
981 )
983 # Generate a query for any stock items for this part variant tree with non-empty serial numbers
984 if not get_global_setting('SERIAL_NUMBER_GLOBALLY_UNIQUE', False): 984 ↛ 989line 984 didn't jump to line 989 because the condition on line 984 was always true
985 # Serial numbers are unique across part trees
986 stock = stock.filter(part__tree_id=self.tree_id)
988 # There are no matching StockItem objects (skip further tests)
989 if not stock.exists(): 989 ↛ 993line 989 didn't jump to line 993 because the condition on line 989 was always true
990 return None
992 # Sort in descending order
993 stock = stock.order_by('-serial_int', '-serial', '-pk')
995 # Return the first serial value
996 return stock[0].serial
998 def get_next_serial_number(self):
999 """Return the 'next' serial number in sequence."""
1000 sn = self.get_latest_serial_number()
1002 return InvenTree.helpers.increment_serial_number(sn, self)
1004 @property
1005 def full_name(self) -> str:
1006 """Format a 'full name' for this Part based on the format PART_NAME_FORMAT defined in InvenTree settings."""
1007 return part_helpers.render_part_full_name(self)
1009 def get_absolute_url(self):
1010 """Return the web URL for viewing this part."""
1011 return helpers.pui_url(f'/part/{self.id}')
1013 def validate_unique(self, exclude=None):
1014 """Validate that this Part instance is 'unique'.
1016 Uniqueness is checked across the following (case insensitive) fields:
1017 - Name
1018 - IPN
1019 - Revision
1021 e.g. there can exist multiple parts with the same name, but only if
1022 they have a different revision or internal part number.
1023 """
1024 super().validate_unique(exclude)
1026 # User can decide whether duplicate IPN (Internal Part Number) values are allowed
1027 allow_duplicate_ipn = get_global_setting('PART_ALLOW_DUPLICATE_IPN')
1029 # Raise an error if an IPN is set, and it is a duplicate
1030 if self.IPN and not allow_duplicate_ipn: 1030 ↛ 1031line 1030 didn't jump to line 1031 because the condition on line 1030 was never true
1031 parts = Part.objects.filter(IPN__iexact=self.IPN)
1032 parts = parts.exclude(pk=self.pk)
1034 if parts.exists():
1035 raise ValidationError({
1036 'IPN': _('Duplicate IPN not allowed in part settings')
1037 })
1039 if ( 1039 ↛ 1049line 1039 didn't jump to line 1049 because the condition on line 1039 was never true
1040 self.revision_of
1041 and self.revision
1042 and (
1043 Part.objects
1044 .exclude(pk=self.pk)
1045 .filter(revision_of=self.revision_of, revision=self.revision)
1046 .exists()
1047 )
1048 ):
1049 raise ValidationError(_('Duplicate part revision already exists.'))
1051 # Ensure unique across (Name, revision, IPN) (as specified)
1052 if (self.revision or self.IPN) and ( 1052 ↛ 1058line 1052 didn't jump to line 1058 because the condition on line 1052 was never true
1053 Part.objects
1054 .exclude(pk=self.pk)
1055 .filter(name=self.name, revision=self.revision, IPN=self.IPN)
1056 .exists()
1057 ):
1058 raise ValidationError(
1059 _('Part with this Name, IPN and Revision already exists.')
1060 )
1062 def clean(self):
1063 """Perform cleaning operations for the Part model.
1065 - Check if the PartCategory is not structural
1067 - Update trackable status:
1068 If this part is trackable, and it is used in the BOM
1069 for a parent part which is *not* trackable,
1070 then we will force the parent part to be trackable.
1071 """
1072 if self.category is not None and self.category.structural: 1072 ↛ 1073line 1072 didn't jump to line 1073 because the condition on line 1072 was never true
1073 raise ValidationError({
1074 'category': _('Parts cannot be assigned to structural part categories!')
1075 })
1077 # Check the 'revision' and 'revision_of' fields
1078 self.validate_revision()
1080 super().clean()
1082 # Strip IPN field
1083 if type(self.IPN) is str: 1083 ↛ 1087line 1083 didn't jump to line 1087 because the condition on line 1083 was always true
1084 self.IPN = self.IPN.strip()
1086 # Run custom validation for the IPN field
1087 self.validate_ipn()
1089 # Run custom validation for the name field
1090 self.validate_name()
1092 if self.pk:
1093 # Only run if the part already exists in the database
1094 self.ensure_trackable()
1096 def ensure_trackable(self):
1097 """Ensure that trackable is set correctly downstream."""
1098 if self.trackable: 1098 ↛ 1099line 1098 didn't jump to line 1099 because the condition on line 1098 was never true
1099 for part in self.get_used_in():
1100 if not part.trackable:
1101 part.trackable = True
1102 part.clean()
1103 part.save()
1105 name = models.CharField(
1106 max_length=100, blank=False, help_text=_('Part name'), verbose_name=_('Name')
1107 )
1109 is_template = models.BooleanField(
1110 default=part_settings.part_template_default,
1111 verbose_name=_('Is Template'),
1112 help_text=_('Is this part a template part?'),
1113 )
1115 variant_of = models.ForeignKey(
1116 'part.Part',
1117 related_name='variants',
1118 null=True,
1119 blank=True,
1120 limit_choices_to={'is_template': True},
1121 on_delete=models.SET_NULL,
1122 help_text=_('Is this part a variant of another part?'),
1123 verbose_name=_('Variant Of'),
1124 )
1126 description = models.CharField(
1127 max_length=250,
1128 blank=True,
1129 verbose_name=_('Description'),
1130 help_text=_('Part description (optional)'),
1131 )
1133 keywords = models.CharField(
1134 max_length=250,
1135 blank=True,
1136 null=True,
1137 verbose_name=_('Keywords'),
1138 help_text=_('Part keywords to improve visibility in search results'),
1139 )
1141 category = TreeForeignKey(
1142 PartCategory,
1143 related_name='parts',
1144 null=True,
1145 blank=True,
1146 on_delete=models.DO_NOTHING,
1147 verbose_name=_('Category'),
1148 help_text=_('Part category'),
1149 )
1151 IPN = models.CharField(
1152 max_length=100,
1153 blank=True,
1154 null=True,
1155 verbose_name=_('IPN'),
1156 help_text=_('Internal Part Number'),
1157 )
1159 revision = models.CharField(
1160 max_length=100,
1161 blank=True,
1162 null=True,
1163 help_text=_('Part revision or version number'),
1164 verbose_name=_('Revision'),
1165 )
1167 revision_of = models.ForeignKey(
1168 'part.Part',
1169 related_name='revisions',
1170 null=True,
1171 blank=True,
1172 on_delete=models.SET_NULL,
1173 help_text=_('Is this part a revision of another part?'),
1174 verbose_name=_('Revision Of'),
1175 )
1177 link = InvenTreeURLField(
1178 blank=True,
1179 null=True,
1180 verbose_name=_('Link'),
1181 help_text=_('Link to external URL'),
1182 max_length=2000,
1183 )
1185 default_location = TreeForeignKey(
1186 'stock.StockLocation',
1187 on_delete=models.SET_NULL,
1188 blank=True,
1189 null=True,
1190 help_text=_('Where is this item normally stored?'),
1191 related_name='default_parts',
1192 verbose_name=_('Default Location'),
1193 )
1195 def get_default_location(self):
1196 """Get the default location for a Part (may be None).
1198 If the Part does not specify a default location,
1199 look at the Category this part is in.
1200 The PartCategory object may also specify a default stock location
1201 """
1202 if self.default_location:
1203 return self.default_location
1204 elif self.category:
1205 # Traverse up the category tree until we find a default location
1206 cats = self.category.get_ancestors(ascending=True, include_self=True)
1208 for cat in cats:
1209 if cat.default_location:
1210 return cat.default_location
1212 # Default case - no default category found
1213 return None
1215 @property
1216 def default_supplier(self):
1217 """Return the default (primary) SupplierPart for this Part.
1219 This function is included for backwards compatibility,
1220 as the 'Part' model used to have a 'default_supplier' field which was a ForeignKey to SupplierPart.
1221 """
1222 return self.supplier_parts.filter(primary=True).first()
1224 default_expiry = models.PositiveIntegerField(
1225 default=0,
1226 validators=[MinValueValidator(0)],
1227 verbose_name=_('Default Expiry'),
1228 help_text=_('Expiry time (in days) for stock items of this part'),
1229 )
1231 minimum_stock = models.DecimalField(
1232 max_digits=19,
1233 decimal_places=6,
1234 default=0,
1235 validators=[MinValueValidator(0)],
1236 verbose_name=_('Minimum Stock'),
1237 help_text=_('Minimum allowed stock level'),
1238 )
1240 maximum_stock = models.DecimalField(
1241 max_digits=19,
1242 decimal_places=6,
1243 default=0,
1244 validators=[MinValueValidator(0)],
1245 verbose_name=_('Maximum Stock'),
1246 help_text=_('Maximum allowed stock level'),
1247 )
1249 units = models.CharField(
1250 max_length=20,
1251 default='',
1252 blank=True,
1253 null=True,
1254 verbose_name=_('Units'),
1255 help_text=_('Units of measure for this part'),
1256 validators=[validators.validate_physical_units],
1257 )
1259 assembly = models.BooleanField(
1260 default=part_settings.part_assembly_default,
1261 verbose_name=_('Assembly'),
1262 help_text=_('Can this part be built from other parts?'),
1263 )
1265 component = models.BooleanField(
1266 default=part_settings.part_component_default,
1267 verbose_name=_('Component'),
1268 help_text=_('Can this part be used to build other parts?'),
1269 )
1271 trackable = models.BooleanField(
1272 default=part_settings.part_trackable_default,
1273 verbose_name=_('Trackable'),
1274 help_text=_('Does this part have tracking for unique items?'),
1275 )
1277 testable = models.BooleanField(
1278 default=False,
1279 verbose_name=_('Testable'),
1280 help_text=_('Can this part have test results recorded against it?'),
1281 )
1283 purchaseable = models.BooleanField(
1284 default=part_settings.part_purchaseable_default,
1285 verbose_name=_('Purchaseable'),
1286 help_text=_('Can this part be purchased from external suppliers?'),
1287 )
1289 salable = models.BooleanField(
1290 default=part_settings.part_salable_default,
1291 verbose_name=_('Salable'),
1292 help_text=_('Can this part be sold to customers?'),
1293 )
1295 active = models.BooleanField(
1296 default=True, verbose_name=_('Active'), help_text=_('Is this part active?')
1297 )
1299 locked = models.BooleanField(
1300 default=False,
1301 verbose_name=_('Locked'),
1302 help_text=_('Locked parts cannot be edited'),
1303 )
1305 virtual = models.BooleanField(
1306 default=part_settings.part_virtual_default,
1307 verbose_name=_('Virtual'),
1308 help_text=_('Is this a virtual part, such as a software product or license?'),
1309 )
1311 bom_validated = models.BooleanField(
1312 default=False,
1313 verbose_name=_('BOM Validated'),
1314 help_text=_('Is the BOM for this part valid?'),
1315 )
1317 bom_checksum = models.CharField(
1318 max_length=128,
1319 blank=True,
1320 verbose_name=_('BOM checksum'),
1321 help_text=_('Stored BOM checksum'),
1322 )
1324 bom_checked_by = models.ForeignKey(
1325 User,
1326 on_delete=models.SET_NULL,
1327 blank=True,
1328 null=True,
1329 verbose_name=_('BOM checked by'),
1330 related_name='boms_checked',
1331 )
1333 bom_checked_date = models.DateField(
1334 blank=True, null=True, verbose_name=_('BOM checked date')
1335 )
1337 creation_date = models.DateField(
1338 auto_now_add=True,
1339 editable=False,
1340 blank=True,
1341 null=True,
1342 verbose_name=_('Creation Date'),
1343 )
1345 creation_user = models.ForeignKey(
1346 User,
1347 on_delete=models.SET_NULL,
1348 blank=True,
1349 null=True,
1350 verbose_name=_('Creation User'),
1351 related_name='parts_created',
1352 )
1354 responsible_owner = models.ForeignKey(
1355 users.models.Owner,
1356 on_delete=models.SET_NULL,
1357 blank=True,
1358 null=True,
1359 verbose_name=_('Responsible'),
1360 help_text=_('Owner responsible for this part'),
1361 related_name='parts_responsible',
1362 )
1364 @property
1365 def category_path(self):
1366 """Return the category path of this Part instance."""
1367 if self.category:
1368 return self.category.pathstring
1369 return ''
1371 @property
1372 def available_stock(self):
1373 """Return the total available stock.
1375 - This subtracts stock which is already allocated to builds
1376 """
1377 total = self.total_stock
1378 total -= self.allocation_count()
1380 return max(total, 0)
1382 def requiring_build_orders(self, include_variants: bool = True):
1383 """Return list of outstanding build orders which require this part.
1385 Arguments:
1386 include_variants: If True, include variants of this part in the calculation
1387 """
1388 # List parts that this part is required for
1390 if include_variants: 1390 ↛ 1394line 1390 didn't jump to line 1394 because the condition on line 1390 was always true
1391 # If we are including variants, get all parts in the variant tree
1392 parts = list(self.get_descendants(include_self=True))
1393 else:
1394 parts = [self]
1396 used_in_parts = set()
1398 for part in parts:
1399 # Get all assemblies which use this part
1400 used_in_parts.update(part.get_used_in())
1402 # Now, get a list of outstanding build orders which require this part
1403 builds = BuildModels.Build.objects.filter(
1404 part__in=list(used_in_parts), status__in=BuildStatusGroups.ACTIVE_CODES
1405 )
1407 return builds
1409 def required_build_order_quantity(self, include_variants: bool = True):
1410 """Return the quantity of this part required for active build orders.
1412 Arguments:
1413 include_variants: If True, include variants of this part in the calculation
1414 """
1415 # List active build orders which reference this part
1416 builds = self.requiring_build_orders(include_variants=include_variants)
1418 quantity = 0
1420 if include_variants: 1420 ↛ 1423line 1420 didn't jump to line 1423 because the condition on line 1420 was always true
1421 matching_parts = list(self.get_descendants(include_self=True))
1422 else:
1423 matching_parts = [self]
1425 # Cache the BOM items that we query
1426 # Keep a dict of part ID to BOM items
1427 cached_bom_items: dict = {}
1429 for build in builds: 1429 ↛ 1430line 1429 didn't jump to line 1430 because the loop on line 1429 never started
1430 if build.part.pk not in cached_bom_items:
1431 # Get the BOM items for this part
1432 bom_items = build.part.get_bom_items().filter(
1433 sub_part__in=matching_parts
1434 )
1435 cached_bom_items[build.part.pk] = bom_items
1436 else:
1437 bom_items = cached_bom_items[build.part.pk]
1439 # Match BOM item to build
1440 for bom_item in bom_items:
1441 build_line = build.build_lines.filter(bom_item=bom_item).first()
1443 if not build_line:
1444 continue
1446 line_quantity = max(0, build_line.quantity - build_line.consumed)
1447 quantity += line_quantity
1449 return quantity
1451 def requiring_sales_orders(self, include_variants: bool = True):
1452 """Return a list of sales orders which require this part.
1454 Arguments:
1455 include_variants: If True, include variants of this part in the calculation
1456 """
1457 orders = set()
1459 if include_variants:
1460 parts = list(self.get_descendants(include_self=True))
1461 else:
1462 parts = [self]
1464 # Get a list of line items for open orders which match this part
1465 open_lines = OrderModels.SalesOrderLineItem.objects.filter(
1466 order__status__in=SalesOrderStatusGroups.OPEN, part__in=parts
1467 )
1469 for line in open_lines:
1470 orders.add(line.order)
1472 return orders
1474 def required_sales_order_quantity(self, include_variants: bool = True):
1475 """Return the quantity of this part required for active sales orders.
1477 Arguments:
1478 include_variants: If True, include variants of this part in the calculation
1479 """
1480 if include_variants: 1480 ↛ 1483line 1480 didn't jump to line 1483 because the condition on line 1480 was always true
1481 parts = list(self.get_descendants(include_self=True))
1482 else:
1483 parts = [self]
1485 # Get a list of line items for open orders which match this part
1486 open_lines = OrderModels.SalesOrderLineItem.objects.filter(
1487 order__status__in=SalesOrderStatusGroups.OPEN, part__in=parts
1488 )
1490 quantity = 0
1492 for line in open_lines: 1492 ↛ 1495line 1492 didn't jump to line 1495 because the loop on line 1492 never started
1493 # Determine the quantity "remaining" to be shipped out
1495 if not line:
1496 continue
1498 remaining = max(line.quantity - line.shipped, 0)
1499 quantity += remaining
1501 return quantity
1503 def required_order_quantity(self, include_variants: bool = True):
1504 """Return total required to fulfil orders."""
1505 return self.required_build_order_quantity(
1506 include_variants=include_variants
1507 ) + self.required_sales_order_quantity(include_variants=include_variants)
1509 @property
1510 def quantity_to_order(self):
1511 """Return the quantity needing to be ordered for this part.
1513 Here, an "order" could be one of:
1514 - Build Order
1515 - Sales Order
1517 To work out how many we need to order:
1519 Stock on hand = self.total_stock
1520 Required for orders = self.required_order_quantity()
1521 Currently on order = self.on_order
1522 Currently building = self.quantity_being_built
1523 """
1524 # Total requirement
1525 required = self.required_order_quantity()
1527 # Subtract stock levels
1528 required -= max(self.total_stock, self.minimum_stock)
1530 # Subtract quantity on order
1531 required -= self.on_order
1533 # Subtract quantity being built
1534 required -= self.quantity_being_built
1536 return max(required, 0)
1538 @property
1539 def net_stock(self):
1540 """Return the 'net' stock.
1542 It takes into account:
1543 - Stock on hand (total_stock)
1544 - Stock on order (on_order)
1545 - Stock allocated (allocation_count)
1547 This number (unlike 'available_stock') can be negative.
1548 """
1549 return self.total_stock - self.allocation_count() + self.on_order
1551 def get_subscribers(
1552 self, include_variants: bool = True, include_categories: bool = True
1553 ) -> list[User]:
1554 """Return a list of users who are 'subscribed' to this part.
1556 Arguments:
1557 include_variants: If True, include users who are subscribed to a variant part
1558 include_categories: If True, include users who are subscribed to the category
1560 Returns:
1561 list[User]: A list of users who are subscribed to this part
1563 A user may 'subscribe' to this part in the following ways:
1565 a) Subscribing to the part instance directly
1566 b) Subscribing to a template part "above" this part (if it is a variant)
1567 c) Subscribing to the part category that this part belongs to
1568 d) Subscribing to a parent category of the category in c)
1569 """
1570 subscribers = set()
1572 # Start by looking at direct subscriptions to a Part model
1573 queryset = PartStar.objects.all()
1575 if include_variants:
1576 queryset = queryset.filter(part__in=self.get_ancestors(include_self=True))
1577 else:
1578 queryset = queryset.filter(part=self)
1580 for star in queryset:
1581 subscribers.add(star.user)
1583 if include_categories and self.category:
1584 for sub in self.category.get_subscribers():
1585 subscribers.add(sub)
1587 return list(subscribers)
1589 def is_starred_by(self, user, **kwargs):
1590 """Return True if the specified user subscribes to this part."""
1591 return user in self.get_subscribers(**kwargs)
1593 def set_starred(self, user, status, **kwargs):
1594 """Set the "subscription" status of this Part against the specified user."""
1595 if not user:
1596 return
1598 # Already subscribed?
1599 if self.is_starred_by(user, **kwargs) == status:
1600 return
1602 if status:
1603 PartStar.objects.create(part=self, user=user)
1604 else:
1605 # Note that this won't actually stop the user being subscribed,
1606 # if the user is subscribed to a parent part or category
1607 PartStar.objects.filter(part=self, user=user).delete()
1609 @property
1610 def can_build(self):
1611 """Return the number of units that can be build with available stock."""
1612 import part.filters
1614 # If this part does NOT have a BOM, result is simply the currently available stock
1615 if not self.has_bom: 1615 ↛ 1619line 1615 didn't jump to line 1619 because the condition on line 1615 was always true
1616 return 0
1618 # Ignore virtual parts when calculating the "can_build" quantity
1619 queryset = self.get_bom_items(include_virtual=False)
1621 # Ignore 'consumable' BOM items for this calculation
1622 queryset = queryset.filter(consumable=False)
1624 # Annotate the queryset with the 'can_build' quantity
1625 queryset = part.filters.annotate_bom_item_can_build(queryset)
1627 can_build_quantity = None
1629 for value in queryset.values_list('can_build', flat=True):
1630 if can_build_quantity is None:
1631 can_build_quantity = value
1632 else:
1633 can_build_quantity = min(can_build_quantity, value)
1635 if can_build_quantity is None:
1636 # No BOM items, or no items which can be built
1637 return 0
1639 return int(max(can_build_quantity, 0))
1641 @property
1642 def active_builds(self):
1643 """Return a list of outstanding builds.
1645 Builds marked as 'complete' or 'cancelled' are ignored
1646 """
1647 return self.builds.filter(status__in=BuildStatusGroups.ACTIVE_CODES)
1649 @property
1650 def quantity_being_built(self, include_variants: bool = True):
1651 """Return the current number of parts currently being built.
1653 Arguments:
1654 include_variants: If True, include variants of this part in the calculation
1656 Note: This is the total quantity of Build orders, *not* the number of build outputs.
1657 In this fashion, it is the "projected" quantity of builds
1658 """
1659 builds = BuildModels.Build.objects.filter(
1660 status__in=BuildStatusGroups.ACTIVE_CODES
1661 )
1663 if include_variants: 1663 ↛ 1668line 1663 didn't jump to line 1668 because the condition on line 1663 was always true
1664 # If we are including variants, get all parts in the variant tree
1665 builds = builds.filter(part__in=self.get_descendants(include_self=True))
1666 else:
1667 # Only look at this part
1668 builds = builds.filter(part=self)
1670 quantity = 0
1672 for build in builds: 1672 ↛ 1674line 1672 didn't jump to line 1674 because the loop on line 1672 never started
1673 # The remaining items in the build
1674 quantity += build.remaining
1676 return quantity
1678 @property
1679 def quantity_in_production(self, include_variants: bool = True):
1680 """Quantity of this part currently actively in production.
1682 Arguments:
1683 include_variants: If True, include variants of this part in the calculation
1685 Note: This may return a different value to `quantity_being_built`
1686 """
1687 quantity = 0
1689 items = StockModels.StockItem.objects.filter(
1690 is_building=True, build__status__in=BuildStatusGroups.ACTIVE_CODES
1691 )
1693 if include_variants: 1693 ↛ 1698line 1693 didn't jump to line 1698 because the condition on line 1693 was always true
1694 # If we are including variants, get all parts in the variant tree
1695 items = items.filter(part__in=self.get_descendants(include_self=True))
1696 else:
1697 # Only look at this part
1698 items = items.filter(part=self)
1700 for item in items: 1700 ↛ 1702line 1700 didn't jump to line 1702 because the loop on line 1700 never started
1701 # The remaining items in the build
1702 quantity += item.quantity
1704 return quantity
1706 def build_order_allocations(self, **kwargs):
1707 """Return all 'BuildItem' objects which allocate this part to Build objects."""
1708 include_variants = kwargs.get('include_variants', True)
1710 queryset = BuildModels.BuildItem.objects.all()
1712 if include_variants: 1712 ↛ 1716line 1712 didn't jump to line 1716 because the condition on line 1712 was always true
1713 variants = self.get_descendants(include_self=True)
1714 queryset = queryset.filter(stock_item__part__in=variants)
1715 else:
1716 queryset = queryset.filter(stock_item__part=self)
1718 return queryset
1720 def build_order_allocation_count(self, **kwargs):
1721 """Return the total amount of this part allocated to build orders."""
1722 query = self.build_order_allocations(**kwargs).aggregate(
1723 total=Coalesce(
1724 Sum('quantity', output_field=models.DecimalField()),
1725 0,
1726 output_field=models.DecimalField(),
1727 )
1728 )
1730 return query['total']
1732 def sales_order_allocations(self, **kwargs):
1733 """Return all sales-order-allocation objects which allocate this part to a SalesOrder."""
1734 include_variants = kwargs.get('include_variants', True)
1736 queryset = OrderModels.SalesOrderAllocation.objects.all()
1738 if include_variants: 1738 ↛ 1744line 1738 didn't jump to line 1744 because the condition on line 1738 was always true
1739 # Include allocations for all variants
1740 variants = self.get_descendants(include_self=True)
1741 queryset = queryset.filter(item__part__in=variants)
1742 else:
1743 # Only look at this part
1744 queryset = queryset.filter(item__part=self)
1746 # Default behaviour is to only return *pending* allocations
1747 pending = kwargs.get('pending', True)
1749 if pending is True: 1749 ↛ 1755line 1749 didn't jump to line 1755 because the condition on line 1749 was always true
1750 # Look only for 'open' orders which have not shipped
1751 queryset = queryset.filter(
1752 line__order__status__in=SalesOrderStatusGroups.OPEN,
1753 shipment__shipment_date=None,
1754 )
1755 elif pending is False:
1756 # Look only for 'closed' orders or orders which have shipped
1757 queryset = queryset.exclude(
1758 line__order__status__in=SalesOrderStatusGroups.OPEN,
1759 shipment__shipment_date=None,
1760 )
1762 return queryset
1764 def sales_order_allocation_count(self, **kwargs):
1765 """Return the total quantity of this part allocated to sales orders."""
1766 query = self.sales_order_allocations(**kwargs).aggregate(
1767 total=Coalesce(
1768 Sum('quantity', output_field=models.DecimalField()),
1769 0,
1770 output_field=models.DecimalField(),
1771 )
1772 )
1774 return query['total']
1776 def transfer_order_allocations(self, **kwargs):
1777 """Return all transfer-order-allocation objects which allocate this part to a TransferOrder."""
1778 include_variants = kwargs.get('include_variants', True)
1780 queryset = OrderModels.TransferOrderAllocation.objects.all()
1782 if include_variants:
1783 # Include allocations for all variants
1784 variants = self.get_descendants(include_self=True)
1785 queryset = queryset.filter(item__part__in=variants)
1786 else:
1787 # Only look at this part
1788 queryset = queryset.filter(item__part=self)
1790 # Default behaviour is to only return *pending* allocations
1791 pending = kwargs.get('pending', True)
1793 if pending is True:
1794 # Look only for 'open' orders
1795 queryset = queryset.filter(
1796 line__order__status__in=TransferOrderStatusGroups.OPEN
1797 )
1798 elif pending is False:
1799 # Look only for 'closed' orders
1800 queryset = queryset.exclude(
1801 line__order__status__in=TransferOrderStatusGroups.OPEN
1802 )
1804 return queryset
1806 def transfer_order_allocation_count(self, **kwargs):
1807 """Return the total quantity of this part allocated to transfer orders."""
1808 query = self.transfer_order_allocations(**kwargs).aggregate(
1809 total=Coalesce(
1810 Sum('quantity', output_field=models.DecimalField()),
1811 0,
1812 output_field=models.DecimalField(),
1813 )
1814 )
1816 return query['total']
1818 def allocation_count(self, **kwargs):
1819 """Return the total quantity of stock allocated for this part, against build orders, sales orders, and transfer orders."""
1820 if self.id is None: 1820 ↛ 1822line 1820 didn't jump to line 1822 because the condition on line 1820 was never true
1821 # If this instance has not been saved, foreign-key lookups will fail
1822 return 0
1824 return sum([
1825 self.build_order_allocation_count(**kwargs),
1826 self.sales_order_allocation_count(**kwargs),
1827 # For now, stock allocated to a transfer order will not impact its availability
1828 # self.transfer_order_allocation_count(**kwargs),
1829 ])
1831 def stock_entries(
1832 self, include_variants=True, include_external=True, in_stock=None, location=None
1833 ):
1834 """Return all stock entries for this Part.
1836 Arguments:
1837 include_variants: If True, include stock entries for all part variants
1838 include_external: If True, include stock entries which are in 'external' locations
1839 in_stock: If True, filter by stock entries which are 'in stock'
1840 location: If set, filter by stock entries in the specified location
1841 """
1842 if include_variants: 1842 ↛ 1847line 1842 didn't jump to line 1847 because the condition on line 1842 was always true
1843 query = StockModels.StockItem.objects.filter(
1844 part__in=self.get_descendants(include_self=True)
1845 )
1846 else:
1847 query = self.stock_items
1849 if in_stock is True: 1849 ↛ 1851line 1849 didn't jump to line 1851 because the condition on line 1849 was always true
1850 query = query.filter(StockModels.StockItem.IN_STOCK_FILTER)
1851 elif in_stock is False:
1852 query = query.exclude(StockModels.StockItem.IN_STOCK_FILTER)
1854 if include_external is False: 1854 ↛ 1856line 1854 didn't jump to line 1856 because the condition on line 1854 was never true
1855 # Exclude stock entries which are not 'internal'
1856 query = query.filter(location__external=False)
1858 if location: 1858 ↛ 1859line 1858 didn't jump to line 1859 because the condition on line 1858 was never true
1859 locations = location.get_descendants(include_self=True)
1860 query = query.filter(location__in=locations)
1862 return query
1864 def get_stock_count(self, include_variants=True):
1865 """Return the total "in stock" count for this part."""
1866 entries = self.stock_entries(in_stock=True, include_variants=include_variants)
1868 query = entries.aggregate(t=Coalesce(Sum('quantity'), Decimal(0)))
1870 return query['t']
1872 @property
1873 def total_stock(self):
1874 """Return the total stock quantity for this part.
1876 - Part may be stored in multiple locations
1877 - If this part is a "template" (variants exist) then these are counted too
1878 """
1879 return self.get_stock_count(include_variants=True)
1881 def get_bom_item_filter(
1882 self, include_inherited: bool = True, include_virtual: bool = True
1883 ):
1884 """Returns a query filter for all BOM items associated with this Part.
1886 Arguments:
1887 include_inherited: If True, include BomItem entries defined for parent parts
1888 include_virtual: If True, include BomItem entries which are virtual
1890 There are some considerations:
1892 a) BOM items can be defined against *this* part
1893 b) BOM items can be inherited from a *parent* part
1895 We will construct a filter to grab *all* the BOM items!
1897 Note: This does *not* return a queryset, it returns a Q object,
1898 which can be used by some other query operation!
1899 Because we want to keep our code DRY!
1900 """
1901 bom_filter = Q(part=self)
1903 if include_inherited:
1904 # We wish to include parent parts
1906 parents = self.get_ancestors(include_self=False)
1908 # There are parents available
1909 if parents.exists(): 1909 ↛ 1910line 1909 didn't jump to line 1910 because the condition on line 1909 was never true
1910 parent_filter = Q(part__in=parents, inherited=True)
1912 # OR the filters together
1913 bom_filter |= parent_filter
1915 if not include_virtual: 1915 ↛ 1916line 1915 didn't jump to line 1916 because the condition on line 1915 was never true
1916 bom_filter &= Q(sub_part__virtual=False)
1918 return bom_filter
1920 def get_bom_items(
1921 self, include_inherited: bool = True, include_virtual: bool = True
1922 ) -> QuerySet[BomItem]:
1923 """Return a queryset containing all BOM items for this part.
1925 Arguments:
1926 include_inherited (bool): If set, include BomItem entries defined for parent parts
1927 include_virtual (bool): If set, include BomItem entries which are virtual parts
1928 """
1929 queryset = BomItem.objects.filter(
1930 self.get_bom_item_filter(
1931 include_inherited=include_inherited, include_virtual=include_virtual
1932 )
1933 )
1935 return queryset.prefetch_related('part', 'sub_part')
1937 def get_installed_part_options(
1938 self, include_inherited: bool = True, include_variants: bool = True
1939 ):
1940 """Return a set of all Parts which can be "installed" into this part, based on the BOM.
1942 Arguments:
1943 include_inherited (bool): If set, include BomItem entries defined for parent parts
1944 include_variants (bool): If set, include variant parts for BomItems which allow variants
1945 """
1946 parts = set()
1948 for bom_item in self.get_bom_items(include_inherited=include_inherited):
1949 if include_variants and bom_item.allow_variants:
1950 for part in bom_item.sub_part.get_descendants(include_self=True):
1951 parts.add(part)
1952 else:
1953 parts.add(bom_item.sub_part)
1955 return parts
1957 def get_used_in_bom_item_filter(
1958 self, include_variants=True, include_substitutes=True
1959 ):
1960 """Return a BomItem queryset which returns all BomItem instances which refer to *this* part.
1962 As the BOM allocation logic is somewhat complicated, there are some considerations:
1964 A) This part may be directly specified in a BomItem instance
1965 B) This part may be a *variant* of a part which is directly specified in a BomItem instance
1966 C) This part may be a *substitute* for a part which is directly specified in a BomItem instance
1968 So we construct a query for each case, and combine them...
1969 """
1970 # Cache all *parent* parts
1971 try:
1972 parents = self.get_ancestors(include_self=False)
1973 except ValueError:
1974 # If get_ancestors() fails, then this part is not saved yet
1975 parents = []
1977 # Case A: This part is directly specified in a BomItem (we always use this case)
1978 query = Q(sub_part=self)
1980 if include_variants: 1980 ↛ 1985line 1980 didn't jump to line 1985 because the condition on line 1980 was always true
1981 # Case B: This part is a *variant* of a part which is specified in a BomItem which allows variants
1982 query |= Q(allow_variants=True, sub_part__in=parents)
1984 # Case C: This part is a *substitute* of a part which is directly specified in a BomItem
1985 if include_substitutes: 1985 ↛ 1991line 1985 didn't jump to line 1991 because the condition on line 1985 was always true
1986 # Grab a list of BomItem substitutes which reference this part
1987 substitutes = self.substitute_items.all()
1989 query |= Q(pk__in=[substitute.bom_item.pk for substitute in substitutes])
1991 return query
1993 def get_used_in(self, include_inherited=True, include_substitutes=True):
1994 """Return a list containing all parts this part is used in.
1996 Includes consideration of inherited BOMs
1997 """
1998 # Grab a queryset of all BomItem objects which "require" this part
1999 bom_items = BomItem.objects.filter(
2000 self.get_used_in_bom_item_filter(include_substitutes=include_substitutes)
2001 )
2003 # Iterate through the returned items and construct a set of
2004 parts = set()
2006 for bom_item in bom_items: 2006 ↛ 2007line 2006 didn't jump to line 2007 because the loop on line 2006 never started
2007 if bom_item.part in parts:
2008 continue
2010 parts.add(bom_item.part)
2012 # Include inherited BOMs?
2013 if include_inherited and bom_item.inherited:
2014 try:
2015 descendants = bom_item.part.get_descendants(include_self=False)
2016 except ValueError:
2017 # This part is not saved yet
2018 descendants = []
2020 for variant in descendants:
2021 parts.add(variant)
2023 return list(parts)
2025 @property
2026 def has_bom(self):
2027 """Return True if this Part instance has any BOM items."""
2028 return self.get_bom_items().exists()
2030 def get_trackable_parts(self):
2031 """Return a queryset of all trackable parts in the BOM for this part."""
2032 queryset = self.get_bom_items()
2033 queryset = queryset.filter(sub_part__trackable=True)
2035 return queryset
2037 @property
2038 def has_trackable_parts(self):
2039 """Return True if any parts linked in the Bill of Materials are trackable.
2041 This is important when building the part.
2042 """
2043 return self.get_trackable_parts().exists()
2045 @property
2046 def bom_count(self):
2047 """Return the number of items contained in the BOM for this part."""
2048 return self.get_bom_items().count()
2050 @property
2051 def used_in_count(self):
2052 """Return the number of part BOMs that this part appears in."""
2053 return len(self.get_used_in())
2055 def get_bom_hash(self):
2056 """Return a checksum hash for the BOM for this part.
2058 Used to determine if the BOM has changed (and needs to be signed off!)
2059 The hash is calculated by hashing each line item in the BOM. Returns a string representation of a hash object which can be compared with a stored value
2060 """
2061 result_hash = hashlib.md5(str(self.id).encode())
2063 # List *all* BOM items (including inherited ones!)
2064 bom_items = self.get_bom_items().all().prefetch_related('part', 'sub_part')
2066 for item in bom_items:
2067 result_hash.update(str(item.get_item_hash()).encode())
2069 return str(result_hash.digest())
2071 def is_bom_valid(self) -> bool:
2072 """Check if the BOM is 'valid'.
2074 To be "valid", the part must:
2075 - Have a stored "bom_checksum" value
2076 - The stored "bom_checksum" must match the calculated checksum.
2078 Returns:
2079 bool: True if the BOM is valid, False otherwise
2080 """
2081 if not self.bom_checksum or not self.bom_checked_date:
2082 # If there is no BOM checksum, then the BOM is not valid
2083 return False
2085 return self.get_bom_hash() == self.bom_checksum
2087 @transaction.atomic
2088 def validate_bom(self, user, valid: bool = True):
2089 """Validate the BOM (mark the BOM as validated by the given User.
2091 Arguments:
2092 user: User who is validating the BOM
2093 valid: If True, mark the BOM as valid (default=True)
2095 - Calculates and stores the hash for the BOM
2096 - Saves the current date and the checking user
2097 """
2098 # Validate each line item, ignoring inherited ones
2099 bom_items = self.get_bom_items(include_inherited=False).prefetch_related(
2100 'part', 'sub_part'
2101 )
2103 if valid: 2103 ↛ 2104line 2103 didn't jump to line 2104 because the condition on line 2103 was never true
2104 for item in bom_items:
2105 item.validate_hash(valid=True)
2107 self.bom_validated = valid
2108 self.bom_checksum = self.get_bom_hash() if valid else ''
2109 self.bom_checked_by = user
2110 self.bom_checked_date = InvenTree.helpers.current_date()
2112 self.save()
2114 @transaction.atomic
2115 def clear_bom(self):
2116 """Clear the BOM items for the part (delete all BOM lines).
2118 Note: Does *NOT* delete inherited BOM items!
2119 """
2120 import part.tasks as part_tasks
2122 self.bom_items.all().delete()
2124 # Offload task to re-validate the BOM for this assembly
2125 InvenTree.tasks.offload_task(part_tasks.check_bom_valid, self.pk, group='part')
2127 def getRequiredParts(self, recursive=False, parts=None):
2128 """Return a list of parts required to make this part (i.e. BOM items).
2130 Args:
2131 recursive: If True iterate down through sub-assemblies
2132 parts: Set of parts already found (to prevent recursion issues)
2133 """
2134 if parts is None:
2135 parts = set()
2137 bom_items = self.get_bom_items()
2139 for bom_item in bom_items:
2140 sub_part = bom_item.sub_part
2142 if sub_part not in parts:
2143 parts.add(sub_part)
2145 if recursive:
2146 sub_part.getRequiredParts(recursive=True, parts=parts)
2148 return parts
2150 @property
2151 def supplier_count(self):
2152 """Return the number of supplier parts available for this part."""
2153 return self.supplier_parts.count()
2155 def update_pricing(self):
2156 """Recalculate cached pricing for this Part instance."""
2157 self.pricing.update_pricing()
2159 @property
2160 def pricing(self):
2161 """Return the PartPricing information for this Part instance.
2163 If there is no PartPricing database entry defined for this Part,
2164 it will first be created, and then returned.
2165 """
2166 try:
2167 pricing = PartPricing.objects.get(part=self)
2168 except PartPricing.DoesNotExist:
2169 pricing = PartPricing(part=self)
2171 return pricing
2173 def schedule_pricing_update(
2174 self, create: bool = False, force: bool = False, refresh: bool = True
2175 ):
2176 """Helper function to schedule a pricing update.
2178 Importantly, catches any errors which may occur during deletion of related objects,
2179 in particular due to post_delete signals.
2181 Ref: https://github.com/inventree/InvenTree/pull/3986
2183 Arguments:
2184 create: Whether or not a new PartPricing object should be created if it does not already exist
2185 force: If True, force the pricing to be updated even auto pricing is disabled
2186 refresh: If True, refresh the PartPricing object from the database
2187 """
2188 if not force and not get_global_setting(
2189 'PRICING_AUTO_UPDATE', backup_value=True
2190 ):
2191 return
2193 if refresh:
2194 try:
2195 self.refresh_from_db()
2196 except Part.DoesNotExist:
2197 return
2199 try:
2200 pricing = self.pricing
2202 if create or pricing.pk:
2203 pricing.schedule_for_update(refresh=refresh)
2204 except IntegrityError:
2205 # If this part instance has been deleted,
2206 # some post-delete or post-save signals may still be fired
2207 # which can cause issues down the track
2208 pass
2210 def get_price_info(self, quantity=1, buy=True, bom=True, internal=False):
2211 """Return a simplified pricing string for this part.
2213 Args:
2214 quantity: Number of units to calculate price for
2215 buy: Include supplier pricing (default = True)
2216 bom: Include BOM pricing (default = True)
2217 internal: Include internal pricing (default = False)
2218 """
2219 price_range = self.get_price_range(quantity, buy, bom, internal)
2221 if price_range is None:
2222 return None
2224 min_price, max_price = price_range
2226 if min_price == max_price:
2227 return min_price
2229 min_price = normalize(min_price)
2230 max_price = normalize(max_price)
2232 return f'{min_price} - {max_price}'
2234 def get_supplier_price_range(self, quantity=1):
2235 """Return the supplier price range of this part.
2237 Actions:
2238 - Checks if there is any supplier pricing information associated with this Part
2239 - Iterate through available supplier pricing and select (min, max)
2240 - Returns tuple of (min, max)
2242 Arguments:
2243 quantity: Quantity at which to calculate price (default=1)
2245 Returns: (min, max) tuple or (None, None) if no supplier pricing available
2246 """
2247 min_price = None
2248 max_price = None
2250 for supplier in self.supplier_parts.all():
2251 price = supplier.get_price(quantity)
2253 if price is None:
2254 continue
2256 if min_price is None or price < min_price:
2257 min_price = price
2259 if max_price is None or price > max_price:
2260 max_price = price
2262 if min_price is None or max_price is None:
2263 return None
2265 min_price = normalize(min_price)
2266 max_price = normalize(max_price)
2268 return (min_price, max_price)
2270 def get_bom_price_range(self, quantity=1, internal=False, purchase=False):
2271 """Return the price range of the BOM for this part.
2273 Adds the minimum price for all components in the BOM.
2274 Note: If the BOM contains items without pricing information,
2275 these items cannot be included in the BOM!
2276 """
2277 min_price = None
2278 max_price = None
2280 for item in self.get_bom_items().select_related('sub_part'):
2281 if item.sub_part.pk == self.pk:
2282 logger.warning('WARNING: BomItem ID %s contains itself in BOM', item.pk)
2283 continue
2285 q = Decimal(quantity)
2286 i = Decimal(item.quantity)
2288 prices = item.sub_part.get_price_range(
2289 q * i, internal=internal, purchase=purchase
2290 )
2292 if prices is None:
2293 continue
2295 low, high = prices
2297 if min_price is None:
2298 min_price = 0
2300 if max_price is None:
2301 max_price = 0
2303 min_price += low
2304 max_price += high
2306 if min_price is None or max_price is None:
2307 return None
2309 min_price = normalize(min_price)
2310 max_price = normalize(max_price)
2312 return (min_price, max_price)
2314 def get_price_range(
2315 self, quantity=1, buy=True, bom=True, internal=False, purchase=False
2316 ):
2317 """Return the price range for this part.
2319 This price can be either:
2320 - Supplier price (if purchased from suppliers)
2321 - BOM price (if built from other parts)
2322 - Internal price (if set for the part)
2323 - Purchase price (if set for the part)
2325 Returns:
2326 Minimum of the supplier, BOM, internal or purchase price. If no pricing available, returns None
2327 """
2328 # only get internal price if set and should be used
2329 if internal and self.has_internal_price_breaks:
2330 internal_price = self.get_internal_price(quantity)
2331 return internal_price, internal_price
2333 # only get purchase price if set and should be used
2334 if purchase:
2335 purchase_price = self.get_purchase_price(quantity)
2336 if purchase_price:
2337 return purchase_price
2339 buy_price_range = self.get_supplier_price_range(quantity) if buy else None
2340 bom_price_range = (
2341 self.get_bom_price_range(quantity, internal=internal) if bom else None
2342 )
2344 if buy_price_range is None:
2345 return bom_price_range
2347 elif bom_price_range is None:
2348 return buy_price_range
2349 return (
2350 min(buy_price_range[0], bom_price_range[0]),
2351 max(buy_price_range[1], bom_price_range[1]),
2352 )
2354 base_cost = models.DecimalField(
2355 max_digits=19,
2356 decimal_places=6,
2357 default=0,
2358 validators=[MinValueValidator(0)],
2359 verbose_name=_('base cost'),
2360 help_text=_('Minimum charge (e.g. stocking fee)'),
2361 )
2363 multiple = models.PositiveIntegerField(
2364 default=1,
2365 validators=[MinValueValidator(1)],
2366 verbose_name=_('multiple'),
2367 help_text=_('Sell multiple'),
2368 )
2370 get_price = common.currency.get_price
2372 @property
2373 def has_price_breaks(self):
2374 """Return True if this part has sale price breaks."""
2375 return self.price_breaks.exists()
2377 @property
2378 def price_breaks(self):
2379 """Return the associated price breaks in the correct order."""
2380 return self.salepricebreaks.order_by('quantity').all()
2382 @property
2383 def unit_pricing(self):
2384 """Returns the price of this Part at quantity=1."""
2385 return self.get_price(1)
2387 def add_price_break(self, quantity, price):
2388 """Create a new price break for this part.
2390 Args:
2391 quantity: Numerical quantity
2392 price: Must be a Money object
2393 """
2394 # Check if a price break at that quantity already exists...
2395 if self.price_breaks.filter(quantity=quantity, part=self.pk).exists():
2396 return
2398 PartSellPriceBreak.objects.create(part=self, quantity=quantity, price=price)
2400 def get_internal_price(self, quantity, moq=True, multiples=True, currency=None):
2401 """Return the internal price of this Part at the specified quantity."""
2402 return common.currency.get_price(
2403 self, quantity, moq, multiples, currency, break_name='internal_price_breaks'
2404 )
2406 @property
2407 def has_internal_price_breaks(self):
2408 """Return True if this Part has internal pricing information."""
2409 return self.internal_price_breaks.exists()
2411 @property
2412 def internal_price_breaks(self):
2413 """Return the associated price breaks in the correct order."""
2414 return self.internalpricebreaks.order_by('quantity').all()
2416 def get_purchase_price(self, quantity):
2417 """Calculate the purchase price for this part at the specified quantity.
2419 - Looks at available supplier pricing data
2420 - Calculates the price base on the closest price point
2421 """
2422 currency = currency_code_default()
2423 try:
2424 prices = [
2425 convert_money(item.purchase_price, currency).amount
2426 for item in self.stock_items.all()
2427 if item.purchase_price
2428 ]
2429 except MissingRate:
2430 prices = None
2432 if prices:
2433 return min(prices) * quantity, max(prices) * quantity
2435 return None
2437 @transaction.atomic
2438 def copy_bom_from(self, other, clear: bool = True, **kwargs):
2439 """Copy the BOM from another part.
2441 Args:
2442 other: The part to copy the BOM from
2443 clear (bool, optional): Remove existing BOM items first. Defaults to True.
2444 """
2445 # Ignore if the other part is actually this part?
2446 if other == self:
2447 return
2449 if clear:
2450 # Remove existing BOM items
2451 # Note: Inherited BOM items are *not* deleted!
2452 self.bom_items.all().delete()
2454 # List of "ancestor" parts above this one
2455 my_ancestors = self.get_ancestors(include_self=False)
2457 raise_error = not kwargs.get('skip_invalid', True)
2459 include_inherited = kwargs.get('include_inherited', False)
2461 # Should substitute parts be duplicated?
2462 copy_substitutes = kwargs.get('copy_substitutes', True)
2464 # Copy existing BOM items from another part
2465 # Note: Inherited BOM Items will *not* be duplicated!!
2466 for bom_item in other.get_bom_items(include_inherited=include_inherited).all():
2467 # If this part already has a BomItem pointing to the same sub-part,
2468 # delete that BomItem from this part first!
2470 # Ignore invalid BomItem objects
2471 if not bom_item.part or not bom_item.sub_part:
2472 continue
2474 # Ignore ancestor parts which are inherited
2475 if bom_item.part in my_ancestors and bom_item.inherited:
2476 continue
2478 # Skip (or throw error) if BomItem is not valid
2479 if not bom_item.sub_part.check_add_to_bom(self, raise_error=raise_error):
2480 continue
2482 # Obtain a list of direct substitute parts against this BomItem
2483 substitutes = BomItemSubstitute.objects.filter(bom_item=bom_item)
2485 # Construct a new BOM item
2486 bom_item.part = self
2487 bom_item.pk = None
2489 bom_item.save()
2490 bom_item.refresh_from_db()
2492 if copy_substitutes:
2493 for sub in substitutes:
2494 # Duplicate the substitute (and point to the *new* BomItem object)
2495 sub.pk = None
2496 sub.bom_item = bom_item
2497 sub.save()
2499 @transaction.atomic
2500 def copy_tests_from(self, other: Part, **kwargs) -> None:
2501 """Copy all test templates from another Part instance.
2503 Note: We only copy the direct test templates, not ones inherited from parent parts.
2504 """
2505 templates = []
2506 parts = self.get_ancestors(include_self=True)
2508 # Prevent tests from being created for non-testable parts
2509 if not self.testable:
2510 return
2512 for template in other.test_templates.all():
2513 # Skip if a test template already exists for this part / key combination
2514 if PartTestTemplate.objects.filter(
2515 key=template.key, part__in=parts
2516 ).exists():
2517 continue
2519 template.pk = None
2520 template.part = self
2521 templates.append(template)
2523 if len(templates) > 0:
2524 PartTestTemplate.objects.bulk_create(templates, batch_size=250)
2526 @transaction.atomic
2527 def copy_category_parameters(self, category: PartCategory):
2528 """Copy parameter templates from the specified PartCategory.
2530 This function is normally called when the Part is first created.
2531 """
2532 from common.models import Parameter
2534 categories = category.get_ancestors(include_self=True)
2536 category_templates = PartCategoryParameterTemplate.objects.filter(
2537 category__in=categories
2538 ).order_by('-category__level')
2540 template_ids = set()
2541 parameters = []
2542 content_type = ContentType.objects.get_for_model(Part)
2544 for category_template in category_templates:
2545 # First ensure that the part doesn't have that parameter
2546 if self.parameters_list.filter(
2547 template=category_template.template
2548 ).exists():
2549 continue
2551 # Ensure we do not create duplicate parameters if multiple categories have the same template
2552 if category_template.template.pk in template_ids:
2553 continue
2555 template_ids.add(category_template.template.pk)
2557 parameters.append(
2558 Parameter(
2559 template=category_template.template,
2560 model_type=content_type,
2561 model_id=self.pk,
2562 data=category_template.default_value,
2563 )
2564 )
2566 Parameter.objects.bulk_create(parameters, batch_size=250)
2568 def getTestTemplates(
2569 self, required=None, include_parent: bool = True, enabled=None
2570 ) -> QuerySet[PartTestTemplate]:
2571 """Return a list of all test templates associated with this Part.
2573 These are used for validation of a StockItem.
2576 Args:
2577 required (bool, optional): Filter templates by whether they are required. Defaults to None.
2578 include_parent (bool, optional): Include templates from parent parts. Defaults to True.
2579 enabled (bool, optional): Filter templates by their enabled status. Defaults to None.
2581 Returns:
2582 QuerySet: A queryset of matching test templates.
2583 """
2584 if include_parent:
2585 tests = PartTestTemplate.objects.filter(
2586 part__in=self.get_ancestors(include_self=True)
2587 )
2588 else:
2589 tests = self.test_templates
2591 if required is not None:
2592 tests = tests.filter(required=required)
2594 if enabled is not None:
2595 tests = tests.filter(enabled=enabled)
2597 return tests
2599 def getTestTemplateMap(self, **kwargs):
2600 """Return a map of all test templates associated with this Part."""
2601 templates = {}
2603 for template in self.getTestTemplates(**kwargs):
2604 templates[template.key] = template
2606 return templates
2608 def getRequiredTests(self, include_parent=True, enabled=True):
2609 """Return the tests which are required by this part.
2611 Arguments:
2612 include_parent: If True, include tests which are defined for parent parts
2613 enabled: If set (either True or False), filter by template "enabled" status
2614 """
2615 return self.getTestTemplates(
2616 required=True, enabled=enabled, include_parent=include_parent
2617 )
2619 def sales_orders(self):
2620 """Return a list of sales orders which reference this part."""
2621 orders = []
2623 for line in self.sales_order_line_items.all().prefetch_related('order'):
2624 if line.order not in orders:
2625 orders.append(line.order)
2627 return orders
2629 def purchase_orders(self):
2630 """Return a list of purchase orders which reference this part."""
2631 orders = []
2633 for part in self.supplier_parts.all().prefetch_related(
2634 'purchase_order_line_items'
2635 ):
2636 for order in part.purchase_orders():
2637 if order not in orders:
2638 orders.append(order)
2640 return orders
2642 @property
2643 def on_order(self):
2644 """Return the total number of items on order for this part.
2646 Note that some supplier parts may have a different pack_quantity attribute,
2647 and this needs to be taken into account!
2648 """
2649 from order.models import PurchaseOrderLineItem
2651 quantity = 0
2653 # Find all outstanding PurchaseOrderLineItem objects which reference this part
2654 lines = PurchaseOrderLineItem.objects.filter(
2655 order__status__in=PurchaseOrderStatusGroups.OPEN,
2656 part__part_id=self.pk,
2657 quantity__gt=F('received'),
2658 ).prefetch_related('part')
2660 for line in lines: 2660 ↛ 2661line 2660 didn't jump to line 2661 because the loop on line 2660 never started
2661 remaining = line.quantity - line.received
2663 if remaining > 0:
2664 quantity += line.part.base_quantity(remaining)
2666 return quantity
2668 @property
2669 def has_variants(self):
2670 """Check if this Part object has variants underneath it."""
2671 return self.get_all_variants().exists()
2673 def get_all_variants(self):
2674 """Return all Part object which exist as a variant under this part."""
2675 return self.get_descendants(include_self=False)
2677 @property
2678 def can_convert(self):
2679 """Check if this Part can be "converted" to a different variant.
2681 It can be converted if:
2682 a) It has non-virtual variant parts underneath it
2683 b) It has non-virtual template parts above it
2684 c) It has non-virtual sibling variants
2685 """
2686 return self.get_conversion_options().exists()
2688 def get_conversion_options(self):
2689 """Return options for converting this part to a "variant" within the same tree.
2691 a) Variants underneath this one
2692 b) Immediate parent
2693 c) Siblings
2694 """
2695 parts = []
2697 # Child parts
2698 for child in self.get_descendants(include_self=False):
2699 parts.append(child)
2701 # Immediate parent, and siblings
2702 if self.variant_of:
2703 parts.append(self.variant_of)
2705 siblings = self.get_siblings(include_self=False)
2707 for sib in siblings:
2708 parts.append(sib)
2710 filtered_parts = Part.objects.filter(pk__in=[part.pk for part in parts])
2712 # Ensure this part is not in the queryset, somehow
2713 filtered_parts = filtered_parts.exclude(pk=self.pk)
2715 filtered_parts = filtered_parts.filter(active=True, virtual=False)
2717 return filtered_parts
2719 def get_related_parts(self):
2720 """Return a set of all related parts for this part."""
2721 related_parts = set()
2723 related_parts_1 = self.related_parts_1.filter(part_1__id=self.pk)
2725 related_parts_2 = self.related_parts_2.filter(part_2__id=self.pk)
2727 for related_part in related_parts_1:
2728 # Add to related parts list
2729 related_parts.add(related_part.part_2)
2731 for related_part in related_parts_2:
2732 # Add to related parts list
2733 related_parts.add(related_part.part_1)
2735 return related_parts
2737 @property
2738 def related_count(self):
2739 """Return the number of 'related parts' which point to this Part."""
2740 return len(self.get_related_parts())
2742 def is_part_low_on_stock(self):
2743 """Returns True if the total stock for this part is less than the minimum stock level."""
2744 return self.get_stock_count() < self.minimum_stock
2747@receiver(post_save, sender=Part, dispatch_uid='part_post_save_log')
2748def after_save_part(sender, instance: Part, created, **kwargs):
2749 """Function to be executed after a Part is saved."""
2750 from django.conf import settings
2752 from part import tasks as part_tasks
2754 if instance and not created and not InvenTree.ready.isImportingData():
2755 # Check part stock only if we are *updating* the part (not creating it)
2757 # Run this check in the background
2758 InvenTree.tasks.offload_task(
2759 part_tasks.notify_low_stock_if_required,
2760 instance.pk,
2761 group='notification',
2762 force_async=not settings.TESTING, # Force async unless in testing mode
2763 )
2765 # Schedule a background task to rebuild any supplier parts
2766 InvenTree.tasks.offload_task(
2767 part_tasks.rebuild_supplier_parts,
2768 instance.pk,
2769 force_async=True,
2770 group='part',
2771 )
2774class PartPricing(common.models.MetaMixin):
2775 """Model for caching min/max pricing information for a particular Part.
2777 It is prohibitively expensive to calculate min/max pricing for a part "on the fly".
2778 As min/max pricing does not change very often, we pre-calculate and cache these values.
2780 Whenever pricing is updated, these values are re-calculated and stored.
2782 Pricing information is cached for:
2784 - BOM cost (min / max cost of component items)
2785 - Purchase cost (based on purchase history)
2786 - Internal cost (based on user-specified InternalPriceBreak data)
2787 - Supplier price (based on supplier part data)
2788 - Variant price (min / max cost of any variants)
2789 - Overall best / worst (based on the values listed above)
2790 - Sale price break min / max values
2791 - Historical sale pricing min / max values
2793 Note that this pricing information does not take "quantity" into account:
2794 - This provides a simple min / max pricing range, which is quite valuable in a lot of situations
2795 - Quantity pricing still needs to be calculated
2796 - Quantity pricing can be viewed from the part detail page
2797 - Detailed pricing information is very context specific in any case
2798 """
2800 # When calculating assembly pricing, we limit the depth of the calculation
2801 MAX_PRICING_DEPTH = 50
2803 @property
2804 def is_valid(self):
2805 """Return True if the cached pricing is valid."""
2806 return self.updated is not None
2808 def convert(self, money):
2809 """Attempt to convert money value to default currency.
2811 If a MissingRate error is raised, ignore it and return None
2812 """
2813 if money is None:
2814 return None
2816 target_currency = currency_code_default()
2818 try:
2819 result = convert_money(money, target_currency)
2820 except MissingRate:
2821 logger.warning(
2822 'No currency conversion rate available for %s -> %s',
2823 money.currency,
2824 target_currency,
2825 )
2826 result = None
2828 return result
2830 def schedule_for_update(self, counter: int = 0, refresh: bool = True):
2831 """Schedule this pricing to be updated.
2833 Arguments:
2834 counter: Recursion counter (used to prevent infinite recursion)
2835 refresh: If specified, the PartPricing object will be refreshed from the database
2836 """
2837 import InvenTree.ready
2839 # If importing data, skip pricing update
2840 if InvenTree.ready.isImportingData():
2841 return
2843 # If running data migrations, skip pricing update
2844 if InvenTree.ready.isRunningMigrations():
2845 return
2847 if (
2848 not self.part
2849 or not self.part.pk
2850 or not Part.objects.filter(pk=self.part.pk).exists()
2851 ):
2852 logger.warning(
2853 'Referenced part instance does not exist - skipping pricing update.'
2854 )
2855 return
2857 try:
2858 if refresh and self.pk:
2859 self.refresh_from_db()
2860 except (PartPricing.DoesNotExist, IntegrityError):
2861 # Error thrown if this PartPricing instance has already been removed
2862 logger.warning(
2863 "Error refreshing PartPricing instance for part '%s'", self.part
2864 )
2865 return
2867 # Ensure that the referenced part still exists in the database
2868 try:
2869 p = self.part
2870 if True: # refresh and p.pk:
2871 p.refresh_from_db()
2872 except IntegrityError:
2873 logger.exception(
2874 "Could not update PartPricing as Part '%s' does not exist", self.part
2875 )
2876 return
2878 if self.scheduled_for_update:
2879 # Ignore if the pricing is already scheduled to be updated
2880 logger.debug('Pricing for %s already scheduled for update - skipping', p)
2881 return
2883 if counter > self.MAX_PRICING_DEPTH:
2884 # Prevent infinite recursion / stack depth issues
2885 logger.debug(
2886 counter, f'Skipping pricing update for {p} - maximum depth exceeded'
2887 )
2888 return
2890 try:
2891 self.scheduled_for_update = True
2892 self.save()
2893 except IntegrityError:
2894 # An IntegrityError here likely indicates that the referenced part has already been deleted
2895 logger.exception(
2896 "Could not save PartPricing for part '%s' to the database", self.part
2897 )
2898 return
2900 import part.tasks as part_tasks
2902 # Pricing calculations are performed in the background,
2903 # unless the TESTING_PRICING flag is set
2904 background = not settings.TESTING or not settings.TESTING_PRICING
2906 # Offload task to update the pricing
2907 # Force async, to prevent running in the foreground (unless in testing mode)
2908 InvenTree.tasks.offload_task(
2909 part_tasks.update_part_pricing,
2910 self,
2911 counter=counter,
2912 force_async=background,
2913 group='pricing',
2914 )
2916 def update_pricing(
2917 self,
2918 counter: int = 0,
2919 cascade: bool = True,
2920 previous_min=None,
2921 previous_max=None,
2922 ):
2923 """Recalculate all cost data for the referenced Part instance.
2925 Arguments:
2926 counter: Recursion counter (used to prevent infinite recursion)
2927 cascade: If True, update pricing for all assemblies and templates which use this part
2928 previous_min: Previous minimum price (used to prevent further updates if unchanged)
2929 previous_max: Previous maximum price (used to prevent further updates if unchanged)
2931 """
2932 # If importing data, skip pricing update
2933 if InvenTree.ready.isImportingData():
2934 return
2936 # If running data migrations, skip pricing update
2937 if InvenTree.ready.isRunningMigrations():
2938 return
2940 if self.pk is not None:
2941 try:
2942 self.refresh_from_db()
2943 except PartPricing.DoesNotExist:
2944 pass
2946 self.update_bom_cost(save=False)
2947 self.update_purchase_cost(save=False)
2948 self.update_internal_cost(save=False)
2949 self.update_supplier_cost(save=False)
2950 self.update_variant_cost(save=False)
2951 self.update_sale_cost(save=False)
2953 # Clear scheduling flag
2954 self.scheduled_for_update = False
2956 # Note: save method calls update_overall_cost
2957 try:
2958 self.save()
2959 except IntegrityError:
2960 # Background worker processes may try to concurrently update
2961 pass
2963 pricing_changed = False
2965 # Without previous pricing data, we assume that the pricing has changed
2966 if previous_min != self.overall_min or previous_max != self.overall_max:
2967 pricing_changed = True
2969 # Update parent assemblies and templates
2970 if pricing_changed and cascade:
2971 self.update_assemblies(counter)
2972 self.update_templates(counter)
2974 def update_assemblies(self, counter: int = 0):
2975 """Schedule updates for any assemblies which use this part."""
2976 # If the linked Part is used in any assemblies, schedule a pricing update for those assemblies
2978 used_in_parts = self.part.get_used_in()
2980 for p in used_in_parts:
2981 p.pricing.schedule_for_update(counter=counter + 1)
2983 def update_templates(self, counter: int = 0):
2984 """Schedule updates for any template parts above this part."""
2985 templates = self.part.get_ancestors(include_self=False)
2987 for p in templates:
2988 p.pricing.schedule_for_update(counter + 1)
2990 def save(self, *args, **kwargs):
2991 """Whenever pricing model is saved, automatically update overall prices."""
2992 # Update the currency which was used to perform the calculation
2993 self.currency = currency_code_default()
2995 try:
2996 self.update_overall_cost()
2997 except Exception:
2998 # If something has happened to the Part model, might throw an error
2999 pass
3001 try:
3002 super().save(*args, **kwargs)
3003 except Exception:
3004 # This error may be thrown if there is already duplicate pricing data
3005 pass
3007 def update_bom_cost(self, save=True):
3008 """Recalculate BOM cost for the referenced Part instance.
3010 Iterate through the Bill of Materials, and calculate cumulative pricing:
3012 cumulative_min: The sum of minimum costs for each line in the BOM
3013 cumulative_max: The sum of maximum costs for each line in the BOM
3015 Note: The cumulative costs are calculated based on the specified default currency
3016 """
3017 if not self.part.assembly:
3018 # Not an assembly - no BOM pricing
3019 self.bom_cost_min = None
3020 self.bom_cost_max = None
3022 if save:
3023 self.save()
3025 # Short circuit - no further operations required
3026 return
3028 currency_code = common.currency.currency_code_default()
3030 cumulative_min = Money(0, currency_code)
3031 cumulative_max = Money(0, currency_code)
3033 any_min_elements = False
3034 any_max_elements = False
3036 for bom_item in self.part.get_bom_items():
3037 # Loop through each BOM item which is used to assemble this part
3039 bom_item_min = None
3040 bom_item_max = None
3042 for sub_part in bom_item.get_valid_parts_for_allocation():
3043 # Check each part which *could* be used
3045 if sub_part != bom_item.sub_part and not sub_part.active:
3046 continue
3048 sub_part_pricing = sub_part.pricing
3050 sub_part_min = self.convert(sub_part_pricing.overall_min)
3051 sub_part_max = self.convert(sub_part_pricing.overall_max)
3053 if sub_part_min is not None:
3054 if bom_item_min is None or sub_part_min < bom_item_min:
3055 bom_item_min = sub_part_min
3057 if sub_part_max is not None:
3058 if bom_item_max is None or sub_part_max > bom_item_max:
3059 bom_item_max = sub_part_max
3061 # Update cumulative totals
3062 if bom_item_min is not None:
3063 bom_item_min *= bom_item.quantity
3064 cumulative_min += self.convert(bom_item_min)
3066 any_min_elements = True
3068 if bom_item_max is not None:
3069 bom_item_max *= bom_item.quantity
3070 cumulative_max += self.convert(bom_item_max)
3072 any_max_elements = True
3074 if any_min_elements:
3075 self.bom_cost_min = cumulative_min
3076 else:
3077 self.bom_cost_min = None
3079 if any_max_elements:
3080 self.bom_cost_max = cumulative_max
3081 else:
3082 self.bom_cost_max = None
3084 if save:
3085 self.save()
3087 def update_purchase_cost(self, save=True):
3088 """Recalculate historical purchase cost for the referenced Part instance.
3090 Purchase history only takes into account "completed" purchase orders.
3091 """
3092 # Find all line items for completed orders which reference this part
3093 line_items = OrderModels.PurchaseOrderLineItem.objects.filter(
3094 order__status=PurchaseOrderStatus.COMPLETE.value,
3095 received__gt=0,
3096 part__part=self.part,
3097 )
3099 # Exclude line items which do not have an associated price
3100 line_items = line_items.exclude(purchase_price=None)
3102 purchase_min = None
3103 purchase_max = None
3105 for line in line_items:
3106 if line.purchase_price is None:
3107 continue
3109 # Take supplier part pack size into account
3110 purchase_cost = self.convert(
3111 line.purchase_price / line.part.pack_quantity_native
3112 )
3114 if purchase_cost is None:
3115 continue
3117 if purchase_min is None or purchase_cost < purchase_min:
3118 purchase_min = purchase_cost
3120 if purchase_max is None or purchase_cost > purchase_max:
3121 purchase_max = purchase_cost
3123 # Also check if manual stock item pricing is included
3124 if get_global_setting('PRICING_USE_STOCK_PRICING', True):
3125 items = self.part.stock_items.all()
3127 # Limit to stock items updated within a certain window
3128 days = int(get_global_setting('PRICING_STOCK_ITEM_AGE_DAYS', 0))
3130 if days > 0:
3131 date_threshold = InvenTree.helpers.current_date() - timedelta(days=days)
3132 items = items.filter(updated__gte=date_threshold)
3134 for item in items:
3135 cost = self.convert(item.purchase_price)
3137 # Skip if the cost could not be converted (for some reason)
3138 if cost is None:
3139 continue
3141 if purchase_min is None or cost < purchase_min:
3142 purchase_min = cost
3144 if purchase_max is None or cost > purchase_max:
3145 purchase_max = cost
3147 self.purchase_cost_min = purchase_min
3148 self.purchase_cost_max = purchase_max
3150 if save:
3151 self.save()
3153 def update_internal_cost(self, save=True):
3154 """Recalculate internal cost for the referenced Part instance."""
3155 min_int_cost = None
3156 max_int_cost = None
3158 if get_global_setting('PART_INTERNAL_PRICE', False):
3159 # Only calculate internal pricing if internal pricing is enabled
3160 for pb in self.part.internalpricebreaks.all():
3161 cost = self.convert(pb.price)
3163 if cost is None:
3164 # Ignore if cost could not be converted for some reason
3165 continue
3167 if min_int_cost is None or cost < min_int_cost:
3168 min_int_cost = cost
3170 if max_int_cost is None or cost > max_int_cost:
3171 max_int_cost = cost
3173 self.internal_cost_min = min_int_cost
3174 self.internal_cost_max = max_int_cost
3176 if save:
3177 self.save()
3179 def update_supplier_cost(self, save=True):
3180 """Recalculate supplier cost for the referenced Part instance.
3182 - The limits are simply the lower and upper bounds of available SupplierPriceBreaks
3183 - We do not take "quantity" into account here
3184 """
3185 min_sup_cost = None
3186 max_sup_cost = None
3188 if self.part.purchaseable:
3189 # Iterate through each available SupplierPart instance
3190 for sp in self.part.supplier_parts.all():
3191 # Iterate through each available SupplierPriceBreak instance
3192 for pb in sp.pricebreaks.all():
3193 if pb.price is None:
3194 continue
3196 # Ensure we take supplier part pack size into account
3197 cost = self.convert(pb.price / sp.pack_quantity_native)
3199 if cost is None:
3200 continue
3202 if min_sup_cost is None or cost < min_sup_cost:
3203 min_sup_cost = cost
3205 if max_sup_cost is None or cost > max_sup_cost:
3206 max_sup_cost = cost
3208 self.supplier_price_min = min_sup_cost
3209 self.supplier_price_max = max_sup_cost
3211 if save:
3212 self.save()
3214 def update_variant_cost(self, save=True):
3215 """Update variant cost values.
3217 Here we track the min/max costs of any variant parts.
3218 """
3219 variant_min = None
3220 variant_max = None
3222 active_only = get_global_setting('PRICING_ACTIVE_VARIANTS', False)
3224 if self.part.is_template:
3225 variants = self.part.get_descendants(include_self=False)
3227 for v in variants:
3228 if active_only and not v.active:
3229 # Ignore inactive variant parts
3230 continue
3232 v_min = self.convert(v.pricing.overall_min)
3233 v_max = self.convert(v.pricing.overall_max)
3235 if v_min is not None:
3236 if variant_min is None or v_min < variant_min:
3237 variant_min = v_min
3239 if v_max is not None:
3240 if variant_max is None or v_max > variant_max:
3241 variant_max = v_max
3243 self.variant_cost_min = variant_min
3244 self.variant_cost_max = variant_max
3246 if save:
3247 self.save()
3249 def update_overall_cost(self):
3250 """Update overall cost values.
3252 Here we simply take the minimum / maximum values of the other calculated fields.
3253 """
3254 overall_min = None
3255 overall_max = None
3257 min_costs = [self.bom_cost_min, self.purchase_cost_min, self.internal_cost_min]
3259 max_costs = [self.bom_cost_max, self.purchase_cost_max, self.internal_cost_max]
3261 purchase_history_override = get_global_setting(
3262 'PRICING_PURCHASE_HISTORY_OVERRIDES_SUPPLIER', False
3263 )
3265 if get_global_setting('PRICING_USE_SUPPLIER_PRICING', True): 3265 ↛ 3273line 3265 didn't jump to line 3273 because the condition on line 3265 was always true
3266 # Add supplier pricing data, *unless* historical pricing information should override
3267 if self.purchase_cost_min is None or not purchase_history_override: 3267 ↛ 3270line 3267 didn't jump to line 3270 because the condition on line 3267 was always true
3268 min_costs.append(self.supplier_price_min)
3270 if self.purchase_cost_max is None or not purchase_history_override: 3270 ↛ 3273line 3270 didn't jump to line 3273 because the condition on line 3270 was always true
3271 max_costs.append(self.supplier_price_max)
3273 if get_global_setting('PRICING_USE_VARIANT_PRICING', True): 3273 ↛ 3279line 3273 didn't jump to line 3279 because the condition on line 3273 was always true
3274 # Include variant pricing in overall calculations
3275 min_costs.append(self.variant_cost_min)
3276 max_costs.append(self.variant_cost_max)
3278 # Calculate overall minimum cost
3279 for cost in min_costs:
3280 if cost is None: 3280 ↛ 3284line 3280 didn't jump to line 3284 because the condition on line 3280 was always true
3281 continue
3283 # Ensure we are working in a common currency
3284 cost = self.convert(cost)
3286 if overall_min is None or cost < overall_min:
3287 overall_min = cost
3289 # Calculate overall maximum cost
3290 for cost in max_costs:
3291 if cost is None: 3291 ↛ 3295line 3291 didn't jump to line 3295 because the condition on line 3291 was always true
3292 continue
3294 # Ensure we are working in a common currency
3295 cost = self.convert(cost)
3297 if overall_max is None or cost > overall_max:
3298 overall_max = cost
3300 if get_global_setting('PART_BOM_USE_INTERNAL_PRICE', False): 3300 ↛ 3302line 3300 didn't jump to line 3302 because the condition on line 3300 was never true
3301 # Check if internal pricing should override other pricing
3302 if self.internal_cost_min is not None:
3303 overall_min = self.internal_cost_min
3305 if self.internal_cost_max is not None:
3306 overall_max = self.internal_cost_max
3308 if self.override_min is not None: 3308 ↛ 3309line 3308 didn't jump to line 3309 because the condition on line 3308 was never true
3309 overall_min = self.convert(self.override_min)
3311 self.overall_min = overall_min
3313 if self.override_max is not None: 3313 ↛ 3314line 3313 didn't jump to line 3314 because the condition on line 3313 was never true
3314 overall_max = self.convert(self.override_max)
3316 self.overall_max = overall_max
3318 def update_sale_cost(self, save=True):
3319 """Recalculate sale cost data."""
3320 # Iterate through the sell price breaks
3321 min_sell_price = None
3322 max_sell_price = None
3324 for pb in self.part.salepricebreaks.all():
3325 cost = self.convert(pb.price)
3327 if cost is None:
3328 continue
3330 if min_sell_price is None or cost < min_sell_price:
3331 min_sell_price = cost
3333 if max_sell_price is None or cost > max_sell_price:
3334 max_sell_price = cost
3336 # Record min/max values
3337 self.sale_price_min = min_sell_price
3338 self.sale_price_max = max_sell_price
3340 min_sell_history = None
3341 max_sell_history = None
3343 # Calculate sale price history too
3344 parts = self.part.get_descendants(include_self=True)
3346 # Find all line items for shipped sales orders which reference this part
3347 line_items = OrderModels.SalesOrderLineItem.objects.filter(
3348 order__status__in=SalesOrderStatusGroups.COMPLETE, part__in=parts
3349 )
3351 # Exclude line items which do not have associated pricing data
3352 line_items = line_items.exclude(sale_price=None)
3354 for line in line_items:
3355 cost = self.convert(line.sale_price)
3357 if cost is None:
3358 continue
3360 if min_sell_history is None or cost < min_sell_history:
3361 min_sell_history = cost
3363 if max_sell_history is None or cost > max_sell_history:
3364 max_sell_history = cost
3366 self.sale_history_min = min_sell_history
3367 self.sale_history_max = max_sell_history
3369 if save:
3370 self.save()
3372 currency = models.CharField(
3373 default=currency_code_default,
3374 max_length=10,
3375 verbose_name=_('Currency'),
3376 help_text=_('Currency used to cache pricing calculations'),
3377 choices=common.currency.currency_code_mappings(),
3378 )
3380 scheduled_for_update = models.BooleanField(default=False)
3382 part = models.OneToOneField(
3383 Part,
3384 on_delete=models.CASCADE,
3385 related_name='pricing_data',
3386 verbose_name=_('Part'),
3387 )
3389 bom_cost_min = InvenTree.fields.InvenTreeModelMoneyField(
3390 null=True,
3391 blank=True,
3392 verbose_name=_('Minimum BOM Cost'),
3393 help_text=_('Minimum cost of component parts'),
3394 )
3396 bom_cost_max = InvenTree.fields.InvenTreeModelMoneyField(
3397 null=True,
3398 blank=True,
3399 verbose_name=_('Maximum BOM Cost'),
3400 help_text=_('Maximum cost of component parts'),
3401 )
3403 purchase_cost_min = InvenTree.fields.InvenTreeModelMoneyField(
3404 null=True,
3405 blank=True,
3406 verbose_name=_('Minimum Purchase Cost'),
3407 help_text=_('Minimum historical purchase cost'),
3408 )
3410 purchase_cost_max = InvenTree.fields.InvenTreeModelMoneyField(
3411 null=True,
3412 blank=True,
3413 verbose_name=_('Maximum Purchase Cost'),
3414 help_text=_('Maximum historical purchase cost'),
3415 )
3417 internal_cost_min = InvenTree.fields.InvenTreeModelMoneyField(
3418 null=True,
3419 blank=True,
3420 verbose_name=_('Minimum Internal Price'),
3421 help_text=_('Minimum cost based on internal price breaks'),
3422 )
3424 internal_cost_max = InvenTree.fields.InvenTreeModelMoneyField(
3425 null=True,
3426 blank=True,
3427 verbose_name=_('Maximum Internal Price'),
3428 help_text=_('Maximum cost based on internal price breaks'),
3429 )
3431 supplier_price_min = InvenTree.fields.InvenTreeModelMoneyField(
3432 null=True,
3433 blank=True,
3434 verbose_name=_('Minimum Supplier Price'),
3435 help_text=_('Minimum price of part from external suppliers'),
3436 )
3438 supplier_price_max = InvenTree.fields.InvenTreeModelMoneyField(
3439 null=True,
3440 blank=True,
3441 verbose_name=_('Maximum Supplier Price'),
3442 help_text=_('Maximum price of part from external suppliers'),
3443 )
3445 variant_cost_min = InvenTree.fields.InvenTreeModelMoneyField(
3446 null=True,
3447 blank=True,
3448 verbose_name=_('Minimum Variant Cost'),
3449 help_text=_('Calculated minimum cost of variant parts'),
3450 )
3452 variant_cost_max = InvenTree.fields.InvenTreeModelMoneyField(
3453 null=True,
3454 blank=True,
3455 verbose_name=_('Maximum Variant Cost'),
3456 help_text=_('Calculated maximum cost of variant parts'),
3457 )
3459 override_min = InvenTree.fields.InvenTreeModelMoneyField(
3460 null=True,
3461 blank=True,
3462 verbose_name=_('Minimum Cost'),
3463 help_text=_('Override minimum cost'),
3464 )
3466 override_max = InvenTree.fields.InvenTreeModelMoneyField(
3467 null=True,
3468 blank=True,
3469 verbose_name=_('Maximum Cost'),
3470 help_text=_('Override maximum cost'),
3471 )
3473 overall_min = InvenTree.fields.InvenTreeModelMoneyField(
3474 null=True,
3475 blank=True,
3476 verbose_name=_('Minimum Cost'),
3477 help_text=_('Calculated overall minimum cost'),
3478 )
3480 overall_max = InvenTree.fields.InvenTreeModelMoneyField(
3481 null=True,
3482 blank=True,
3483 verbose_name=_('Maximum Cost'),
3484 help_text=_('Calculated overall maximum cost'),
3485 )
3487 sale_price_min = InvenTree.fields.InvenTreeModelMoneyField(
3488 null=True,
3489 blank=True,
3490 verbose_name=_('Minimum Sale Price'),
3491 help_text=_('Minimum sale price based on price breaks'),
3492 )
3494 sale_price_max = InvenTree.fields.InvenTreeModelMoneyField(
3495 null=True,
3496 blank=True,
3497 verbose_name=_('Maximum Sale Price'),
3498 help_text=_('Maximum sale price based on price breaks'),
3499 )
3501 sale_history_min = InvenTree.fields.InvenTreeModelMoneyField(
3502 null=True,
3503 blank=True,
3504 verbose_name=_('Minimum Sale Cost'),
3505 help_text=_('Minimum historical sale price'),
3506 )
3508 sale_history_max = InvenTree.fields.InvenTreeModelMoneyField(
3509 null=True,
3510 blank=True,
3511 verbose_name=_('Maximum Sale Cost'),
3512 help_text=_('Maximum historical sale price'),
3513 )
3516class PartStocktake(models.Model):
3517 """Model representing a 'stock history' entry for a particular Part.
3519 A 'stocktake' is a representative count of available stock:
3520 - Performed on a given date
3521 - Records quantity of part in stock (across multiple stock items)
3522 - Records estimated value of "stock on hand"
3523 """
3525 part = models.ForeignKey(
3526 Part,
3527 on_delete=models.CASCADE,
3528 related_name='stocktakes',
3529 verbose_name=_('Part'),
3530 help_text=_('Part for stocktake'),
3531 )
3533 item_count = models.IntegerField(
3534 default=1,
3535 verbose_name=_('Item Count'),
3536 help_text=_('Number of individual stock entries at time of stocktake'),
3537 )
3539 quantity = models.DecimalField(
3540 max_digits=19,
3541 decimal_places=5,
3542 validators=[MinValueValidator(0)],
3543 verbose_name=_('Quantity'),
3544 help_text=_('Total available stock at time of stocktake'),
3545 )
3547 date = models.DateField(
3548 verbose_name=_('Date'),
3549 help_text=_('Date stocktake was performed'),
3550 auto_now_add=True,
3551 )
3553 cost_min = InvenTree.fields.InvenTreeModelMoneyField(
3554 null=True,
3555 blank=True,
3556 verbose_name=_('Minimum Stock Cost'),
3557 help_text=_('Estimated minimum cost of stock on hand'),
3558 )
3560 cost_max = InvenTree.fields.InvenTreeModelMoneyField(
3561 null=True,
3562 blank=True,
3563 verbose_name=_('Maximum Stock Cost'),
3564 help_text=_('Estimated maximum cost of stock on hand'),
3565 )
3568class PartSellPriceBreak(common.models.PriceBreak):
3569 """Represents a price break for selling this part."""
3571 class Meta:
3572 """Metaclass providing extra model definition."""
3574 verbose_name = _('Part Sale Price Break')
3575 unique_together = ('part', 'quantity')
3577 @staticmethod
3578 def get_api_url():
3579 """Return the list API endpoint URL associated with the PartSellPriceBreak model."""
3580 return reverse('api-part-sale-price-list')
3582 part = models.ForeignKey(
3583 Part,
3584 on_delete=models.CASCADE,
3585 related_name='salepricebreaks',
3586 limit_choices_to={'salable': True},
3587 verbose_name=_('Part'),
3588 )
3591class PartInternalPriceBreak(common.models.PriceBreak):
3592 """Represents a price break for internally selling this part."""
3594 class Meta:
3595 """Metaclass providing extra model definition."""
3597 unique_together = ('part', 'quantity')
3599 @staticmethod
3600 def get_api_url():
3601 """Return the list API endpoint URL associated with the PartInternalPriceBreak model."""
3602 return reverse('api-part-internal-price-list')
3604 part = models.ForeignKey(
3605 Part,
3606 on_delete=models.CASCADE,
3607 related_name='internalpricebreaks',
3608 verbose_name=_('Part'),
3609 )
3612class PartStar(models.Model):
3613 """A PartStar object creates a subscription relationship between a User and a Part.
3615 It is used to designate a Part as 'subscribed' for a given User.
3617 Attributes:
3618 part: Link to a Part object
3619 user: Link to a User object
3620 """
3622 class Meta:
3623 """Metaclass providing extra model definition."""
3625 unique_together = ['part', 'user']
3627 part = models.ForeignKey(
3628 Part,
3629 on_delete=models.CASCADE,
3630 verbose_name=_('Part'),
3631 related_name='starred_users',
3632 )
3634 user = models.ForeignKey(
3635 User,
3636 on_delete=models.CASCADE,
3637 verbose_name=_('User'),
3638 related_name='starred_parts',
3639 )
3642class PartCategoryStar(models.Model):
3643 """A PartCategoryStar creates a subscription relationship between a User and a PartCategory.
3645 Attributes:
3646 category: Link to a PartCategory object
3647 user: Link to a User object
3648 """
3650 class Meta:
3651 """Metaclass providing extra model definition."""
3653 unique_together = ['category', 'user']
3655 category = models.ForeignKey(
3656 PartCategory,
3657 on_delete=models.CASCADE,
3658 verbose_name=_('Category'),
3659 related_name='starred_users',
3660 )
3662 user = models.ForeignKey(
3663 User,
3664 on_delete=models.CASCADE,
3665 verbose_name=_('User'),
3666 related_name='starred_categories',
3667 )
3670class PartTestTemplate(InvenTree.models.InvenTreeMetadataModel):
3671 """A PartTestTemplate defines a 'template' for a test which is required to be run against a StockItem (an instance of the Part).
3673 The test template applies "recursively" to part variants, allowing tests to be
3674 defined in a hierarchy.
3676 Test names are simply strings, rather than enforcing any sort of structure or pattern.
3677 It is up to the user to determine what tests are defined (and how they are run).
3679 To enable generation of unique lookup-keys for each test, there are some validation tests
3680 run on the model (refer to the validate_unique function).
3681 """
3683 IMPORT_ID_FIELDS = ['key']
3685 class Meta:
3686 """Metaclass options for the PartTestTemplate model."""
3688 verbose_name = _('Part Test Template')
3690 def __str__(self):
3691 """Format a string representation of this PartTestTemplate."""
3692 return ' | '.join([self.part.name, self.test_name])
3694 @staticmethod
3695 def get_api_url():
3696 """Return the list API endpoint URL associated with the PartTestTemplate model."""
3697 return reverse('api-part-test-template-list')
3699 def save(self, *args, **kwargs):
3700 """Enforce 'clean' operation when saving a PartTestTemplate instance."""
3701 self.clean()
3703 super().save(*args, **kwargs)
3705 def clean(self):
3706 """Clean fields for the PartTestTemplate model."""
3707 self.test_name = self.test_name.strip()
3709 self.key = helpers.generateTestKey(self.test_name)
3711 if len(self.key) == 0:
3712 raise ValidationError({
3713 'test_name': _(
3714 'Invalid template name - must include at least one alphanumeric character'
3715 )
3716 })
3718 # Check that 'choices' are in fact valid
3719 if self.choices is None:
3720 self.choices = ''
3721 else:
3722 self.choices = str(self.choices).strip()
3724 if self.choices:
3725 choice_set = set()
3727 for choice in self.choices.split(','):
3728 choice = choice.strip()
3730 # Ignore empty choices
3731 if not choice:
3732 continue
3734 if choice in choice_set:
3735 raise ValidationError({'choices': _('Choices must be unique')})
3737 choice_set.add(choice)
3739 self.validate_unique()
3740 super().clean()
3742 def validate_unique(self, exclude=None):
3743 """Test that this test template is 'unique' within this part tree."""
3744 if not self.part.testable:
3745 raise ValidationError({
3746 'part': _('Test templates can only be created for testable parts')
3747 })
3749 # Check that this test is unique for this part
3750 # (including template parts of which this part is a variant)
3751 parts = self.part.get_ancestors(include_self=True)
3753 tests = PartTestTemplate.objects.filter(key=self.key, part__in=parts).exclude(
3754 pk=self.pk
3755 )
3757 if tests.exists():
3758 raise ValidationError({
3759 'test_name': _(
3760 'Test template with the same key already exists for part'
3761 )
3762 })
3764 super().validate_unique(exclude)
3766 part = models.ForeignKey(
3767 Part,
3768 on_delete=models.CASCADE,
3769 related_name='test_templates',
3770 limit_choices_to={'testable': True},
3771 verbose_name=_('Part'),
3772 )
3774 test_name = models.CharField(
3775 blank=False,
3776 max_length=100,
3777 verbose_name=_('Test Name'),
3778 help_text=_('Enter a name for the test'),
3779 )
3781 key = models.CharField(
3782 blank=True,
3783 max_length=100,
3784 verbose_name=_('Test Key'),
3785 help_text=_('Simplified key for the test'),
3786 )
3788 description = models.CharField(
3789 blank=False,
3790 null=True,
3791 max_length=100,
3792 verbose_name=_('Test Description'),
3793 help_text=_('Enter description for this test'),
3794 )
3796 enabled = models.BooleanField(
3797 default=True, verbose_name=_('Enabled'), help_text=_('Is this test enabled?')
3798 )
3800 required = models.BooleanField(
3801 default=True,
3802 verbose_name=_('Required'),
3803 help_text=_('Is this test required to pass?'),
3804 )
3806 requires_value = models.BooleanField(
3807 default=False,
3808 verbose_name=_('Requires Value'),
3809 help_text=_('Does this test require a value when adding a test result?'),
3810 )
3812 requires_attachment = models.BooleanField(
3813 default=False,
3814 verbose_name=_('Requires Attachment'),
3815 help_text=_(
3816 'Does this test require a file attachment when adding a test result?'
3817 ),
3818 )
3820 choices = models.CharField(
3821 max_length=5000,
3822 verbose_name=_('Choices'),
3823 help_text=_('Valid choices for this test (comma-separated)'),
3824 blank=True,
3825 )
3827 def get_choices(self):
3828 """Return a list of valid choices for this test template."""
3829 if not self.choices:
3830 return []
3832 return [x.strip() for x in self.choices.split(',') if x.strip()]
3835class BomItem(InvenTree.models.MetadataMixin, InvenTree.models.InvenTreeModel):
3836 """A BomItem links a part to its component items.
3838 A part can have a BOM (bill of materials) which defines
3839 which parts are required (and in what quantity) to make it.
3841 Attributes:
3842 part: Link to the parent part (the part that will be produced)
3843 sub_part: Link to the child part (the part that will be consumed)
3844 raw_amount: Raw amount of 'sub_part' consumed to produce one 'part' (can be fractional, or use an associated unit)
3845 quantity: Numerical quantity of 'sub_parts' consumed to produce one 'part'
3846 optional: Boolean field describing if this BomItem is optional
3847 consumable: Boolean field describing if this BomItem is considered a 'consumable'
3848 reference: BOM reference field (e.g. part designators)
3849 setup_quantity: Extra required quantity for a build, to account for setup losses
3850 attrition: Estimated losses for a Build, expressed as a percentage (e.g. '2%')
3851 rounding_multiple: Rounding quantity when calculating the required quantity for a build
3852 note: Note field for this BOM item
3853 checksum: Validation checksum for the particular BOM line item
3854 validated: Boolean field indicating if this BOM item is valid (checksum matches)
3855 inherited: This BomItem can be inherited by the BOMs of variant parts
3856 allow_variants: Stock for part variants can be substituted for this BomItem
3857 """
3859 class Meta:
3860 """Metaclass providing extra model definition."""
3862 verbose_name = _('BOM Item')
3864 def __str__(self):
3865 """Return a string representation of this BomItem instance."""
3866 return f'{decimal2string(self.quantity)} x {self.sub_part.full_name} to make {self.part.full_name}'
3868 @staticmethod
3869 def get_api_url():
3870 """Return the list API endpoint URL associated with the BomItem model."""
3871 return reverse('api-bom-list')
3873 def get_assemblies(self):
3874 """Return a list of assemblies which use this BomItem."""
3875 assemblies = [self.part]
3877 if self.inherited:
3878 assemblies += list(self.part.get_descendants(include_self=False))
3880 return assemblies
3882 def get_valid_parts_for_allocation(
3883 self,
3884 allow_variants: bool = True,
3885 allow_substitutes: bool = True,
3886 allow_inactive: bool = True,
3887 ):
3888 """Return a list of valid parts which can be allocated against this BomItem.
3890 Arguments:
3891 allow_variants: If True, include variants of the sub_part
3892 allow_substitutes: If True, include any directly specified substitute parts
3893 allow_inactive: If True, include inactive parts in the returned list
3895 Includes:
3896 - The referenced sub_part
3897 - Any directly specified substitute parts
3898 - If allow_variants is True, all variants of sub_part
3899 """
3900 # Set of parts we will allow
3901 parts = set()
3903 parts.add(self.sub_part)
3905 # Variant parts (if allowed)
3906 if allow_variants and self.allow_variants:
3907 for variant in self.sub_part.get_descendants(include_self=False):
3908 parts.add(variant)
3910 # Substitute parts
3911 if allow_substitutes:
3912 for sub in self.substitutes.all():
3913 parts.add(sub.part)
3915 # Account for variants of the substitute part (if allowed)
3916 if allow_variants and self.allow_variants:
3917 for sub_variant in sub.part.get_descendants(include_self=False):
3918 parts.add(sub_variant)
3920 valid_parts = []
3922 for p in parts:
3923 # Trackable status must be the same as the sub_part
3924 if p.trackable != self.sub_part.trackable:
3925 continue
3927 # Filter by 'active' status
3928 if not allow_inactive and not p.active:
3929 continue
3931 valid_parts.append(p)
3933 return valid_parts
3935 def is_stock_item_valid(self, stock_item):
3936 """Check if the provided StockItem object is "valid" for assignment against this BomItem."""
3937 return stock_item.part in self.get_valid_parts_for_allocation()
3939 def get_stock_filter(self):
3940 """Return a queryset filter for selecting StockItems which match this BomItem.
3942 - Allow stock from all directly specified substitute parts
3943 - If allow_variants is True, allow all part variants
3944 """
3945 return Q(part__in=self.get_valid_parts_for_allocation())
3947 def set_quantity(self, quantity: Decimal | str | float):
3948 """Update the 'quantity' for this BomItem."""
3949 self.raw_amount = quantity
3950 self.recalculate_quantity()
3952 def recalculate_quantity(self):
3953 """Recalculate the 'quantity' field based on the 'raw_amount' field."""
3954 if self.raw_amount is None or self.raw_amount == '':
3955 self.raw_amount = self.quantity
3957 # Convert from the "raw amount" to a numerical quantity, using the associated unit (if specified)
3958 try:
3959 quantity = InvenTree.conversion.convert_physical_value(
3960 self.raw_amount, self.sub_part.units, strip_units=False
3961 )
3963 if not self.sub_part.units and not InvenTree.conversion.is_dimensionless(
3964 quantity
3965 ):
3966 raise ValidationError({
3967 'raw_amount': _('Invalid quantity - no units specified for part')
3968 })
3970 allow_zero_qty = get_global_setting('PART_BOM_ALLOW_ZERO_QUANTITY', False)
3972 if allow_zero_qty:
3973 if float(quantity.magnitude) < 0:
3974 raise ValidationError({
3975 'raw_amount': _(
3976 'Quantity must be greater than or equal to zero'
3977 )
3978 })
3980 else:
3981 if float(quantity.magnitude) <= 0:
3982 raise ValidationError({
3983 'raw_amount': _('Quantity must be greater than zero')
3984 })
3986 # Normalize the quantity, to maximum 5 decimal places
3987 quantity = Decimal(quantity.magnitude)
3989 except ValidationError as e:
3990 raise ValidationError({'raw_amount': e.messages})
3992 # Ensure that the raw_amount is converted to a Decimal value
3993 # and quantized to a maximum of 5 decimal places (to avoid floating point issues)
3994 try:
3995 self.quantity = Decimal(quantity).quantize(
3996 Decimal('0.00001'), rounding=ROUND_HALF_UP
3997 )
3998 except InvalidOperation:
3999 msg = _('Invalid quantity provided')
4000 raise ValidationError({'quantity': msg, 'raw_amount': msg})
4002 def delete(self):
4003 """Check if this item can be deleted."""
4004 import part.tasks as part_tasks
4006 self.check_part_lock(self.part)
4008 assemblies = self.get_assemblies()
4009 super().delete()
4011 for assembly in assemblies:
4012 # Offload task to update the checksum for this assembly
4013 InvenTree.tasks.offload_task(
4014 part_tasks.check_bom_valid, assembly.pk, group='part'
4015 )
4017 def save(self, *args, **kwargs):
4018 """Enforce 'clean' operation when saving a BomItem instance."""
4019 import part.tasks as part_tasks
4021 self.clean()
4023 check_lock = kwargs.pop('check_lock', True)
4025 if check_lock:
4026 self.check_part_lock(self.part)
4028 db_instance = self.get_db_instance()
4030 # Check if the part was changed
4031 deltas = self.get_field_deltas()
4033 if 'part' in deltas and (old_part := deltas['part'].get('old', None)):
4034 if check_lock:
4035 self.check_part_lock(old_part)
4037 # Update the 'validated' field based on checksum calculation
4038 self.validated = self.is_line_valid
4040 super().save(*args, **kwargs)
4042 # Do we need to recalculate the BOM hash for assemblies?
4043 if not db_instance or any(f in deltas for f in self.hash_fields()):
4044 # If this is a new BomItem, or if any of the fields used to calculate the hash have changed,
4045 # then we need to recalculate the BOM checksum for all assemblies which use this BomItem
4047 assemblies = set()
4049 if db_instance:
4050 # Find all assemblies which use this BomItem *before* we save
4051 assemblies.update(db_instance.get_assemblies())
4053 # Update the set of assemblies to include those which use this BomItem *after* we save
4054 assemblies.update(self.get_assemblies())
4056 for assembly in assemblies:
4057 # Offload task to update the checksum for this assembly
4058 InvenTree.tasks.offload_task(
4059 part_tasks.check_bom_valid, assembly.pk, group='part'
4060 )
4062 def check_part_lock(self, assembly):
4063 """When editing or deleting a BOM item, check if the assembly is locked.
4065 If locked, raise an exception.
4067 Arguments:
4068 assembly: The assembly part
4070 Raises:
4071 ValidationError: If the assembly is locked
4072 """
4073 if not get_global_setting('PART_ENABLE_LOCKING'):
4074 return
4076 if assembly.locked:
4077 raise ValidationError(_('BOM item cannot be modified - assembly is locked'))
4079 # If this BOM item is inherited, check all variants of the assembly
4080 if self.inherited:
4081 for part in assembly.get_descendants(include_self=False):
4082 if part.locked:
4083 raise ValidationError(
4084 _('BOM item cannot be modified - variant assembly is locked')
4085 )
4087 # A link to the parent part
4088 # Each part will get a reverse lookup field 'bom_items'
4089 part = models.ForeignKey(
4090 Part,
4091 on_delete=models.CASCADE,
4092 related_name='bom_items',
4093 verbose_name=_('Part'),
4094 help_text=_('Select parent part'),
4095 limit_choices_to={'assembly': True},
4096 )
4098 # A link to the child item (sub-part)
4099 # Each part will get a reverse lookup field 'used_in'
4100 sub_part = models.ForeignKey(
4101 Part,
4102 on_delete=models.CASCADE,
4103 related_name='used_in',
4104 verbose_name=_('Sub part'),
4105 help_text=_('Select part to be used in BOM'),
4106 limit_choices_to={'component': True},
4107 )
4109 raw_amount = models.CharField(
4110 max_length=25,
4111 verbose_name=_('Amount'),
4112 help_text=_('Amount of sub-part consumed to produce one part'),
4113 blank=False,
4114 null=False,
4115 )
4117 # Native quantity required
4118 quantity = models.DecimalField(
4119 default=1.0,
4120 max_digits=15,
4121 decimal_places=5,
4122 validators=[MinValueValidator(0)],
4123 verbose_name=_('Quantity'),
4124 help_text=_('BOM quantity for this BOM item'),
4125 )
4127 optional = models.BooleanField(
4128 default=False,
4129 verbose_name=_('Optional'),
4130 help_text=_('This BOM item is optional'),
4131 )
4133 consumable = models.BooleanField(
4134 default=False,
4135 verbose_name=_('Consumable'),
4136 help_text=_('This BOM item is consumable (it is not tracked in build orders)'),
4137 )
4139 setup_quantity = models.DecimalField(
4140 default=0,
4141 max_digits=15,
4142 decimal_places=5,
4143 validators=[MinValueValidator(0)],
4144 verbose_name=_('Setup Quantity'),
4145 help_text=_('Extra required quantity for a build, to account for setup losses'),
4146 )
4148 attrition = models.DecimalField(
4149 default=0,
4150 max_digits=6,
4151 decimal_places=3,
4152 validators=[MinValueValidator(0), MaxValueValidator(100)],
4153 verbose_name=_('Attrition'),
4154 help_text=_(
4155 'Estimated attrition for a build, expressed as a percentage (0-100)'
4156 ),
4157 )
4159 rounding_multiple = models.DecimalField(
4160 null=True,
4161 blank=True,
4162 default=None,
4163 max_digits=15,
4164 decimal_places=5,
4165 validators=[MinValueValidator(0)],
4166 verbose_name=_('Rounding Multiple'),
4167 help_text=_(
4168 'Round up required production quantity to nearest multiple of this value'
4169 ),
4170 )
4172 reference = models.CharField(
4173 max_length=5000,
4174 blank=True,
4175 verbose_name=_('Reference'),
4176 help_text=_('BOM item reference'),
4177 )
4179 # Note attached to this BOM line item
4180 note = models.CharField(
4181 max_length=500,
4182 blank=True,
4183 verbose_name=_('Note'),
4184 help_text=_('BOM item notes'),
4185 )
4187 checksum = models.CharField(
4188 max_length=128,
4189 blank=True,
4190 verbose_name=_('Checksum'),
4191 help_text=_('BOM line checksum'),
4192 )
4194 validated = models.BooleanField(
4195 default=False,
4196 verbose_name=_('Validated'),
4197 help_text=_('This BOM item has been validated'),
4198 )
4200 inherited = models.BooleanField(
4201 default=False,
4202 verbose_name=_('Gets inherited'),
4203 help_text=_('This BOM item is inherited by BOMs for variant parts'),
4204 )
4206 allow_variants = models.BooleanField(
4207 default=False,
4208 verbose_name=_('Allow Variants'),
4209 help_text=_('Stock items for variant parts can be used for this BOM item'),
4210 )
4212 def hash_fields(self) -> list[str]:
4213 """Return a list of fields to be used for hashing this BOM item.
4215 These fields are used to calculate the checksum hash of this BOM item.
4216 """
4217 return [
4218 'part',
4219 'part_id',
4220 'sub_part',
4221 'sub_part_id',
4222 'quantity',
4223 'setup_quantity',
4224 'attrition',
4225 'rounding_multiple',
4226 'reference',
4227 'optional',
4228 'inherited',
4229 'consumable',
4230 'allow_variants',
4231 ]
4233 def get_item_hash(self) -> str:
4234 """Calculate the checksum hash of this BOM line item."""
4235 # Seed the hash with the ID of this BOM item
4236 result_hash = hashlib.md5(b'')
4238 for field in self.hash_fields():
4239 # Get the value of the field
4240 value = getattr(self, field, None)
4242 # If the value is None, use an empty string
4243 if value is None:
4244 value = ''
4246 # Normalize decimal values to ensure consistent representation
4247 # These values are only included if they are non-zero
4248 # This is to provide some backwards compatibility from before these fields were added
4249 if value is not None and field in [
4250 'quantity',
4251 'attrition',
4252 'setup_quantity',
4253 'rounding_multiple',
4254 ]:
4255 try:
4256 value = normalize(value)
4258 if not value or value <= 0:
4259 continue
4260 except Exception:
4261 pass
4263 # Update the hash with the string representation of the value
4264 result_hash.update(str(value).encode())
4266 return str(result_hash.digest())
4268 def validate_hash(self, valid=True):
4269 """Mark this item as 'valid' (store the checksum hash).
4271 Args:
4272 valid: If true, validate the hash, otherwise invalidate it (default = True)
4273 """
4274 if valid:
4275 self.checksum = self.get_item_hash()
4276 else:
4277 self.checksum = ''
4279 # Save the BOM item (bypass lock check)
4280 self.save(check_lock=False)
4282 @property
4283 def is_line_valid(self):
4284 """Check if this line item has been validated by the user."""
4285 # Ensure an empty checksum returns False
4286 if len(self.checksum) == 0:
4287 return False
4289 return self.get_item_hash() == self.checksum
4291 def clean(self):
4292 """Check validity of the BomItem model.
4294 Performs model checks beyond simple field validation.
4296 - A part cannot refer to itself in its BOM
4297 - A part cannot refer to a part which refers to it
4299 - If the "sub_part" is trackable, then the "part" must be trackable too!
4300 """
4301 super().clean()
4303 # Recalculate the 'quantity' field based on the 'raw_amount' field
4304 self.recalculate_quantity()
4306 try:
4307 # Check for circular BOM references
4308 if self.sub_part:
4309 self.sub_part.check_add_to_bom(self.part, raise_error=True)
4311 # If the sub_part is 'trackable' then the 'quantity' field must be an integer
4312 if self.sub_part.trackable:
4313 if self.quantity != int(self.quantity):
4314 raise ValidationError({
4315 'quantity': _(
4316 'Quantity must be integer value for trackable parts'
4317 )
4318 })
4320 # Force the upstream part to be trackable if the sub_part is trackable
4321 if not self.part.trackable:
4322 self.part.trackable = True
4323 self.part.clean()
4324 self.part.save()
4325 else:
4326 raise ValidationError({'sub_part': _('Sub part must be specified')})
4327 except Part.DoesNotExist:
4328 raise ValidationError({'sub_part': _('Sub part must be specified')})
4330 def can_build_quantity(self, available_stock: float) -> int:
4331 """Calculate the number of assemblies that can be built with the available stock.
4333 Arguments:
4334 available_stock: The amount of stock available for this BOM item
4336 Returns:
4337 The number of assemblies that can be built with the available stock.
4338 Returns 0 if the available stock is insufficient.
4339 """
4340 # Account for setup quantity
4341 available_stock = Decimal(max(0, available_stock - self.setup_quantity))
4342 quantity_decimal = Decimal(self.quantity)
4343 attrition_decimal = Decimal(self.attrition) / 100
4344 n = quantity_decimal * (1 + attrition_decimal)
4346 if n <= 0:
4347 return 0.0
4349 return int(Decimal(available_stock) / n)
4351 def get_required_quantity(self, build_quantity: float) -> float:
4352 """Calculate the required part quantity, based on the supplied build_quantity.
4354 Arguments:
4355 build_quantity: Number of assemblies to build
4357 Returns:
4358 Production quantity required for this component
4359 """
4360 # Base quantity requirement
4361 required = self.quantity * build_quantity
4363 # Account for attrition
4364 if self.attrition > 0:
4365 try:
4366 # Convert attrition percentage to decimal
4367 attrition = Decimal(self.attrition) / Decimal(100)
4368 required *= 1 + attrition
4369 except Exception:
4370 log_error('bom_item.get_required_quantity')
4372 # Account for setup quantity
4373 if self.setup_quantity > 0:
4374 try:
4375 setup_quantity = Decimal(self.setup_quantity)
4376 required += setup_quantity
4377 except Exception:
4378 log_error('bom_item.get_required_quantity')
4380 # We now have the total requirement
4381 # If a "rounding_multiple" is specified, then round up to the nearest multiple
4382 if self.rounding_multiple and self.rounding_multiple > 0:
4383 try:
4384 round_up = Decimal(self.rounding_multiple)
4385 value = Decimal(required)
4386 value = math.ceil(value / round_up) * round_up
4387 required = float(value)
4388 except InvalidOperation:
4389 log_error('bom_item.get_required_quantity')
4391 return required
4393 @property
4394 def price_range(self, internal=False):
4395 """Return the price-range for this BOM item."""
4396 # get internal price setting
4397 use_internal = get_global_setting('PART_BOM_USE_INTERNAL_PRICE', False)
4398 p_range = self.sub_part.get_price_range(
4399 self.quantity, internal=use_internal and internal
4400 )
4402 if p_range is None:
4403 return p_range
4405 p_min, p_max = p_range
4407 if p_min == p_max:
4408 return decimal2money(p_min)
4410 # Convert to better string representation
4411 p_min = decimal2money(p_min)
4412 p_max = decimal2money(p_max)
4414 return f'{p_min} to {p_max}'
4417@receiver(post_save, sender=BomItem, dispatch_uid='update_bom_build_lines')
4418def update_bom_build_lines(sender, instance, created, **kwargs):
4419 """Update existing build orders when a BomItem is created or edited."""
4420 if InvenTree.ready.canAppAccessDatabase() and not InvenTree.ready.isImportingData():
4421 import build.tasks
4423 InvenTree.tasks.offload_task(
4424 build.tasks.update_build_order_lines, instance.pk, group='build'
4425 )
4428@receiver(post_save, sender=BomItem, dispatch_uid='post_save_bom_item')
4429@receiver(
4430 post_save, sender=PartSellPriceBreak, dispatch_uid='post_save_sale_price_break'
4431)
4432@receiver(
4433 post_save,
4434 sender=PartInternalPriceBreak,
4435 dispatch_uid='post_save_internal_price_break',
4436)
4437def update_pricing_after_edit(sender, instance, created, **kwargs):
4438 """Callback function when a part price break is created or updated."""
4439 # Update part pricing *unless* we are importing data
4440 if (
4441 InvenTree.ready.canAppAccessDatabase(allow_test=settings.TESTING_PRICING)
4442 and not InvenTree.ready.isImportingData()
4443 ):
4444 if instance.part:
4445 instance.part.schedule_pricing_update(create=True)
4448@receiver(post_delete, sender=BomItem, dispatch_uid='post_delete_bom_item')
4449@receiver(
4450 post_delete, sender=PartSellPriceBreak, dispatch_uid='post_delete_sale_price_break'
4451)
4452@receiver(
4453 post_delete,
4454 sender=PartInternalPriceBreak,
4455 dispatch_uid='post_delete_internal_price_break',
4456)
4457def update_pricing_after_delete(sender, instance, **kwargs):
4458 """Callback function when a part price break is deleted."""
4459 # Update part pricing *unless* we are importing data
4460 if (
4461 InvenTree.ready.canAppAccessDatabase(allow_test=settings.TESTING_PRICING)
4462 and not InvenTree.ready.isImportingData()
4463 ):
4464 if instance.part:
4465 instance.part.schedule_pricing_update(create=False)
4468class BomItemSubstitute(InvenTree.models.InvenTreeMetadataModel):
4469 """A BomItemSubstitute provides a specification for alternative parts, which can be used in a bill of materials.
4471 Attributes:
4472 bom_item: Link to the parent BomItem instance
4473 part: The part which can be used as a substitute
4474 """
4476 class Meta:
4477 """Metaclass providing extra model definition."""
4479 verbose_name = _('BOM Item Substitute')
4481 # Prevent duplication of substitute parts
4482 unique_together = ('part', 'bom_item')
4484 def save(self, *args, **kwargs):
4485 """Enforce a full_clean when saving the BomItemSubstitute model."""
4486 self.full_clean()
4488 super().save(*args, **kwargs)
4490 def validate_unique(self, exclude=None):
4491 """Ensure that this BomItemSubstitute is "unique".
4493 Ensure:
4494 - It cannot point to the same "part" as the "sub_part" of the parent "bom_item"
4495 """
4496 super().validate_unique(exclude=exclude)
4498 if self.part == self.bom_item.sub_part:
4499 raise ValidationError({
4500 'part': _('Substitute part cannot be the same as the master part')
4501 })
4503 @staticmethod
4504 def get_api_url():
4505 """Returns the list API endpoint URL associated with this model."""
4506 return reverse('api-bom-substitute-list')
4508 bom_item = models.ForeignKey(
4509 BomItem,
4510 on_delete=models.CASCADE,
4511 related_name='substitutes',
4512 verbose_name=_('BOM Item'),
4513 help_text=_('Parent BOM item'),
4514 )
4516 part = models.ForeignKey(
4517 Part,
4518 on_delete=models.CASCADE,
4519 related_name='substitute_items',
4520 verbose_name=_('Part'),
4521 help_text=_('Substitute part'),
4522 limit_choices_to={'component': True},
4523 )
4526class PartRelated(InvenTree.models.InvenTreeMetadataModel):
4527 """Store and handle related parts (eg. mating connector, crimps, etc.)."""
4529 class Meta:
4530 """Metaclass defines extra model properties."""
4532 unique_together = ('part_1', 'part_2')
4534 part_1 = models.ForeignKey(
4535 Part,
4536 related_name='related_parts_1',
4537 verbose_name=_('Part 1'),
4538 on_delete=models.CASCADE,
4539 )
4541 part_2 = models.ForeignKey(
4542 Part,
4543 related_name='related_parts_2',
4544 on_delete=models.CASCADE,
4545 verbose_name=_('Part 2'),
4546 help_text=_('Select Related Part'),
4547 )
4549 note = models.CharField(
4550 max_length=500,
4551 blank=True,
4552 verbose_name=_('Note'),
4553 help_text=_('Note for this relationship'),
4554 )
4556 def __str__(self):
4557 """Return a string representation of this Part-Part relationship."""
4558 return f'{self.part_1} <--> {self.part_2}'
4560 def save(self, *args, **kwargs):
4561 """Enforce a 'clean' operation when saving a PartRelated instance."""
4562 self.clean()
4563 self.validate_unique()
4564 super().save(*args, **kwargs)
4566 def clean(self):
4567 """Overwrite clean method to check that relation is unique."""
4568 super().clean()
4570 if self.part_1 == self.part_2:
4571 raise ValidationError(
4572 _('Part relationship cannot be created between a part and itself')
4573 )
4575 # Check for inverse relationship
4576 if PartRelated.objects.filter(part_1=self.part_2, part_2=self.part_1).exists():
4577 raise ValidationError(_('Duplicate relationship already exists'))