Coverage for src/backend/InvenTree/InvenTree/models.py: 52%
645 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 17:47 +0000
1"""Generic models which provide extra functionality over base Django model types."""
3from collections.abc import Callable
4from datetime import datetime
5from string import Formatter
6from typing import Any, Optional
8from django.contrib.auth import get_user_model
9from django.contrib.contenttypes.fields import GenericRelation
10from django.contrib.contenttypes.models import ContentType
11from django.core.exceptions import ValidationError
12from django.db import models, transaction
13from django.db.models import QuerySet
14from django.db.models.signals import post_save
15from django.db.transaction import TransactionManagementError
16from django.dispatch import receiver
17from django.urls import resolve, reverse
18from django.urls.exceptions import NoReverseMatch
19from django.utils.translation import gettext_lazy as _
21import structlog
22from django_q.models import Task
23from error_report.models import Error
24from mptt.exceptions import InvalidMove
25from mptt.models import MPTTModel, TreeForeignKey
26from rest_framework.exceptions import PermissionDenied
27from stdimage.models import StdImageField
28from taggit.managers import TaggableManager
30import common.settings
31import InvenTree.exceptions
32import InvenTree.fields
33import InvenTree.format
34import InvenTree.helpers
35import InvenTree.helpers_model
36import InvenTree.sentry
38logger = structlog.get_logger('inventree')
41class DiffMixin:
42 """Mixin which can be used to determine which fields have changed, compared to the instance saved to the database."""
44 def get_db_instance(self):
45 """Return the instance of the object saved in the database.
47 Returns:
48 object: Instance of the object saved in the database
49 """
50 if self.pk:
51 try:
52 return self.__class__.objects.get(pk=self.pk)
53 except self.__class__.DoesNotExist:
54 pass
56 return None
58 def get_field_deltas(self):
59 """Return a dict of field deltas.
61 Compares the current instance with the instance saved in the database,
62 and returns a dict of fields which have changed.
64 Returns:
65 dict: Dict of field deltas
66 """
67 db_instance = self.get_db_instance()
69 if db_instance is None:
70 return {}
72 deltas = {}
74 for field in self._meta.fields:
75 if field.name == 'id':
76 continue
78 if getattr(self, field.name) != getattr(db_instance, field.name):
79 deltas[field.name] = {
80 'old': getattr(db_instance, field.name),
81 'new': getattr(self, field.name),
82 }
84 return deltas
86 def has_field_changed(self, field_name):
87 """Determine if a particular field has changed."""
88 return field_name in self.get_field_deltas()
91class PluginValidationMixin(DiffMixin):
92 """Mixin class which exposes the model instance to plugin validation.
94 Any model class which inherits from this mixin will be exposed to the plugin validation system.
95 """
97 def should_plugin_validate(self):
98 """Return True if this model instance should be validated by plugins.
100 The default implementation returns True, but this can be overridden in the implementing class if required.
101 """
102 from InvenTree.ready import isReadOnlyCommand
104 # Prevent plugin validation when importing or exporting data
105 return not isReadOnlyCommand()
107 def run_plugin_validation(self):
108 """Throw this model against the plugin validation interface."""
109 from plugin import PluginMixinEnum, registry
111 if not self.should_plugin_validate(): 111 ↛ 112line 111 didn't jump to line 112 because the condition on line 111 was never true
112 return
114 deltas = self.get_field_deltas()
116 for plugin in registry.with_mixin(PluginMixinEnum.VALIDATION): 116 ↛ 117line 116 didn't jump to line 117 because the loop on line 116 never started
117 try:
118 if plugin.validate_model_instance(self, deltas=deltas) is True:
119 return
120 except ValidationError as exc:
121 raise exc
122 except Exception:
123 # Log the exception to the database
124 import InvenTree.exceptions
126 InvenTree.exceptions.log_error(
127 'validate_model_instance', plugin=plugin.slug
128 )
129 raise ValidationError(_('Error running plugin validation'))
131 def full_clean(self, *args, **kwargs):
132 """Run plugin validation on full model clean.
134 Note that plugin validation is performed *after* super.full_clean()
135 """
136 super().full_clean(*args, **kwargs)
137 self.run_plugin_validation()
139 def save(self, *args, **kwargs):
140 """Run plugin validation on model save.
142 Note that plugin validation is performed *before* super.save()
143 """
144 self.run_plugin_validation()
145 super().save(*args, **kwargs)
147 def delete(self, *args, **kwargs):
148 """Run plugin validation on model delete.
150 Allows plugins to prevent model instances from being deleted.
152 Note: Each plugin may raise a ValidationError to prevent deletion.
153 """
154 from InvenTree.exceptions import log_error
155 from plugin import PluginMixinEnum, registry
157 if self.should_plugin_validate(): 157 ↛ 168line 157 didn't jump to line 168 because the condition on line 157 was always true
158 for plugin in registry.with_mixin(PluginMixinEnum.VALIDATION): 158 ↛ 159line 158 didn't jump to line 159 because the loop on line 158 never started
159 try:
160 plugin.validate_model_deletion(self)
161 except ValidationError as e:
162 # Plugin might raise a ValidationError to prevent deletion
163 raise e
164 except Exception:
165 log_error('validate_model_deletion', plugin=plugin.slug)
166 continue
168 super().delete(*args, **kwargs)
171class MetadataMixin(models.Model):
172 """Model mixin class which adds a JSON metadata field to a model, for use by any (and all) plugins.
174 The intent of this mixin is to provide a metadata field on a model instance,
175 for plugins to read / modify as required, to store any extra information.
177 The assumptions for models implementing this mixin are:
179 - The internal InvenTree business logic will make no use of this field
180 - Multiple plugins may read / write to this metadata field, and not assume they have sole rights
181 """
183 class Meta:
184 """Meta for MetadataMixin."""
186 abstract = True
188 def save(self, force_insert=False, force_update=False, *args, **kwargs):
189 """Save the model instance, and perform validation on the metadata field."""
190 self.validate_metadata()
191 if len(args) > 0: 191 ↛ 192line 191 didn't jump to line 192 because the condition on line 191 was never true
192 raise TypeError(
193 'save() takes no positional arguments anymore'
194 ) # pragma: no cover
195 super().save(force_insert=force_insert, force_update=force_update, **kwargs)
197 def clean(self, *args, **kwargs):
198 """Perform model validation on the metadata field."""
199 super().clean()
201 self.validate_metadata()
203 def validate_metadata(self):
204 """Validate the metadata field."""
205 # Ensure that the 'metadata' field is a valid dict object
206 if self.metadata is None:
207 self.metadata = {}
209 if type(self.metadata) is not dict: 209 ↛ 210line 209 didn't jump to line 210 because the condition on line 209 was never true
210 raise ValidationError({
211 'metadata': _('Metadata must be a python dict object')
212 })
214 metadata = models.JSONField(
215 blank=True,
216 null=True,
217 verbose_name=_('Plugin Metadata'),
218 help_text=_('JSON metadata field, for use by external plugins'),
219 )
221 def get_metadata(self, key: str, backup_value=None):
222 """Finds metadata for this model instance, using the provided key for lookup.
224 Args:
225 key: String key for requesting metadata. e.g. if a plugin is accessing the metadata, the plugin slug should be used
226 backup_value: Value that should be used if no value is found
228 Returns:
229 Python dict object containing requested metadata. If no matching metadata is found, returns None
230 """
231 if self.metadata is None:
232 return backup_value
234 return self.metadata.get(key, backup_value)
236 def set_metadata(
237 self, key: str, data, commit: bool = True, overwrite: bool = False
238 ):
239 """Save the provided metadata under the provided key.
241 Args:
242 key (str): Key for saving metadata
243 data (Any): Data object to save - must be able to be rendered as a JSON string
244 commit (bool, optional): If true, existing metadata with the provided key will be overwritten. If false, a merge will be attempted. Defaults to True.
245 overwrite (bool): If true, delete existing metadata before adding new value
246 """
247 if overwrite or self.metadata is None: 247 ↛ 248line 247 didn't jump to line 248 because the condition on line 247 was never true
248 self.metadata = {}
250 self.metadata[key] = data
252 if commit: 252 ↛ exitline 252 didn't return from function 'set_metadata' because the condition on line 252 was always true
253 self.save()
256class ReferenceIndexingMixin(models.Model):
257 """A mixin for keeping track of numerical copies of the "reference" field.
259 Here, we attempt to convert a "reference" field value (char) to an integer,
260 for performing fast natural sorting.
262 This requires extra database space (due to the extra table column),
263 but is required as not all supported database backends provide equivalent casting.
265 This mixin adds a field named 'reference_int'.
267 - If the 'reference' field can be cast to an integer, it is stored here
268 - If the 'reference' field *starts* with an integer, it is stored here
269 - Otherwise, we store zero
270 """
272 # Name of the global setting which defines the required reference pattern for this model
273 REFERENCE_PATTERN_SETTING = None
275 class Meta:
276 """Metaclass options. Abstract ensures no database table is created."""
278 abstract = True
280 @classmethod
281 def get_reference_pattern(cls):
282 """Returns the reference pattern associated with this model.
284 This is defined by a global setting object, specified by the REFERENCE_PATTERN_SETTING attribute
285 """
286 # By default, we return an empty string
287 if cls.REFERENCE_PATTERN_SETTING is None: 287 ↛ 288line 287 didn't jump to line 288 because the condition on line 287 was never true
288 return ''
290 return common.settings.get_global_setting(
291 cls.REFERENCE_PATTERN_SETTING, create=False
292 ).strip()
294 @classmethod
295 def get_reference_context(cls):
296 """Generate context data for generating the 'reference' field for this class.
298 - Returns a python dict object which contains the context data for formatting the reference string.
299 - The default implementation provides some default context information
300 - The '?' key is required to accept our wildcard-with-default syntax {?:default}
301 """
302 return {'ref': cls.get_next_reference(), 'date': datetime.now(), '?': '?'}
304 @classmethod
305 def get_most_recent_item(cls):
306 """Return the item which is 'most recent'.
308 In practice, this means the item with the highest reference value
309 """
310 query = cls.objects.all().order_by('-reference_int', '-pk')
312 if query.exists():
313 return query.first()
314 return None
316 @classmethod
317 def get_next_reference(cls):
318 """Return the next available reference value for this particular class."""
319 # Find the "most recent" item
320 latest = cls.get_most_recent_item()
322 if not latest:
323 # No existing items
324 return 1
326 reference = latest.reference.strip
328 try:
329 reference = InvenTree.format.extract_named_group(
330 'ref', reference, cls.get_reference_pattern()
331 )
332 except Exception:
333 # If reference cannot be extracted using the pattern, try just the integer value
334 reference = str(latest.reference_int)
336 # Attempt to perform 'intelligent' incrementing of the reference field
337 incremented = InvenTree.helpers.increment(reference)
339 try:
340 incremented = int(incremented)
341 except ValueError:
342 pass
344 return incremented
346 @classmethod
347 def generate_reference(cls):
348 """Generate the next 'reference' field based on specified pattern."""
350 # Based on https://stackoverflow.com/a/57570269/14488558
351 class ReferenceFormatter(Formatter):
352 def format_field(self, value, format_spec):
353 if isinstance(value, str) and value == '?': 353 ↛ 354line 353 didn't jump to line 354 because the condition on line 353 was never true
354 value = format_spec
355 format_spec = ''
356 return super().format_field(value, format_spec)
358 ref_ptn = cls.get_reference_pattern()
359 ctx = cls.get_reference_context()
360 fmt = ReferenceFormatter()
362 reference = None
364 attempts = set()
366 while reference is None: 366 ↛ 390line 366 didn't jump to line 390 because the condition on line 366 was always true
367 try:
368 ref = fmt.format(ref_ptn, **ctx)
370 if ref in attempts: 370 ↛ 372line 370 didn't jump to line 372 because the condition on line 370 was never true
371 # We are stuck in a loop!
372 reference = ref
373 break
374 else:
375 attempts.add(ref)
377 if cls.objects.filter(reference=ref).exists(): 377 ↛ 379line 377 didn't jump to line 379 because the condition on line 377 was never true
378 # Handle case where we have duplicated an existing reference
379 ctx['ref'] = InvenTree.helpers.increment(ctx['ref'])
380 else:
381 # We have found an 'unused' reference
382 reference = ref
383 break
385 except Exception:
386 # If anything goes wrong, return the most recent reference
387 recent = cls.get_most_recent_item()
388 reference = recent.reference if recent else ''
390 return reference
392 @classmethod
393 def validate_reference_pattern(cls, pattern):
394 """Ensure that the provided pattern is valid."""
395 ctx = cls.get_reference_context()
397 try:
398 info = InvenTree.format.parse_format_string(pattern)
399 except Exception as exc:
400 raise ValidationError({
401 'value': _('Improperly formatted pattern') + ': ' + str(exc)
402 })
404 # Check that only 'allowed' keys are provided
405 for key in info:
406 if key not in ctx:
407 raise ValidationError({
408 'value': _('Unknown format key specified') + f": '{key}'"
409 })
411 # Check that the 'ref' variable is specified
412 if 'ref' not in info:
413 raise ValidationError({
414 'value': _('Missing required format key') + ": 'ref'"
415 })
417 @classmethod
418 def validate_reference_field(cls, value):
419 """Check that the provided 'reference' value matches the requisite pattern."""
420 pattern = cls.get_reference_pattern()
422 value = str(value).strip()
424 if len(value) == 0: 424 ↛ 425line 424 didn't jump to line 425 because the condition on line 424 was never true
425 raise ValidationError(_('Reference field cannot be empty'))
427 # An 'empty' pattern means no further validation is required
428 if not pattern: 428 ↛ 429line 428 didn't jump to line 429 because the condition on line 428 was never true
429 return
431 if not InvenTree.format.validate_string(value, pattern):
432 raise ValidationError(
433 _('Reference must match required pattern') + ': ' + pattern
434 )
436 # Check that the reference field can be rebuild
437 return cls.rebuild_reference_field(value, validate=True)
439 @classmethod
440 def rebuild_reference_field(cls, reference, validate=False):
441 """Extract integer out of reference for sorting.
443 If the 'integer' portion is buried somewhere 'within' the reference,
444 we can first try to extract it using the pattern.
446 Example:
447 reference - BO-123-ABC
448 pattern - BO-{ref}-???
449 extracted - 123
451 If we cannot extract using the pattern for some reason, fallback to the entire reference
452 """
453 try:
454 # Extract named group based on provided pattern
455 reference = InvenTree.format.extract_named_group(
456 'ref', reference, cls.get_reference_pattern()
457 )
458 except Exception:
459 pass
461 reference_int = InvenTree.helpers.extract_int(reference)
463 if validate and reference_int > models.BigIntegerField.MAX_BIGINT: 463 ↛ 464line 463 didn't jump to line 464 because the condition on line 463 was never true
464 raise ValidationError({'reference': _('Reference number is too large')})
466 return reference_int
468 reference_int = models.BigIntegerField(default=0)
471class ContentTypeMixin:
472 """Mixin class which supports retrieval of the ContentType for a model instance."""
474 @classmethod
475 def get_content_type(cls):
476 """Return the ContentType object associated with this model."""
477 from django.contrib.contenttypes.models import ContentType
479 return ContentType.objects.get_for_model(cls)
482class InvenTreeModel(ContentTypeMixin, PluginValidationMixin, models.Model):
483 """Base class for InvenTree models, which provides some common functionality.
485 Includes the following mixins by default:
487 - PluginValidationMixin: Provides a hook for plugins to validate model instances
488 """
490 class Meta:
491 """Metaclass options."""
493 abstract = True
496class InvenTreeMetadataModel(MetadataMixin, InvenTreeModel):
497 """Base class for an InvenTree model which includes a metadata field."""
499 class Meta:
500 """Metaclass options."""
502 abstract = True
505class InvenTreePermissionCheckMixin:
506 """Provides an abstracted class for managing permissions against related fields."""
508 @classmethod
509 def check_related_permission(cls, permission, user) -> bool:
510 """Check if the user has permission to perform the specified action on the attachment.
512 The default implementation runs a permission check against *this* model class,
513 but this can be overridden in the implementing class if required.
515 Arguments:
516 permission: The permission to check (add / change / view / delete)
517 user: The user to check against
519 Returns:
520 bool: True if the user has permission, False otherwise
521 """
522 perm = f'{cls._meta.app_label}.{permission}_{cls._meta.model_name}'
523 return user.has_perm(perm)
526class InvenTreeParameterMixin(InvenTreePermissionCheckMixin, models.Model):
527 """Provides an abstracted class for managing parameters.
529 Links the implementing model to the common.models.Parameter table,
530 and provides the following methods:
531 """
533 class Meta:
534 """Metaclass options for InvenTreeParameterMixin."""
536 abstract = True
538 # Define a reverse relation to the Parameter model
539 parameters_list = GenericRelation(
540 'common.Parameter', content_type_field='model_type', object_id_field='model_id'
541 )
543 @staticmethod
544 def annotate_parameters(queryset: QuerySet) -> QuerySet:
545 """Annotate a queryset with pre-fetched parameters.
547 Args:
548 queryset: Queryset to annotate
550 Returns:
551 Annotated queryset
552 """
553 return queryset.prefetch_related(
554 'parameters_list',
555 'parameters_list__model_type',
556 'parameters_list__updated_by',
557 'parameters_list__template',
558 'parameters_list__template__model_type',
559 )
561 @property
562 def parameters(self) -> QuerySet:
563 """Return a QuerySet containing all the Parameter instances for this model.
565 This will return pre-fetched data if available (i.e. in a serializer context).
566 """
567 # Check the query cache for pre-fetched parameters
568 if cache := getattr(self, '_prefetched_objects_cache', None):
569 if 'parameters_list' in cache: 569 ↛ 572line 569 didn't jump to line 572 because the condition on line 569 was always true
570 return cache['parameters_list']
572 return self.parameters_list.all().prefetch_related('template')
574 def delete(self, *args, **kwargs):
575 """Handle the deletion of a model instance.
577 Before deleting the model instance, delete any associated parameters.
578 """
579 self.parameters_list.all().delete()
580 super().delete(*args, **kwargs)
582 @transaction.atomic
583 def copy_parameters_from(self, other, clear=True, **kwargs):
584 """Copy all parameters from another model instance.
586 Arguments:
587 other: The other model instance to copy parameters from
588 clear: If True, clear existing parameters before copying
589 **kwargs: Additional keyword arguments to pass to the Parameter constructor
590 """
591 import common.models
593 if clear:
594 self.parameters_list.all().delete()
596 parameters = []
598 content_type = ContentType.objects.get_for_model(self.__class__)
600 template_ids = [parameter.template.pk for parameter in other.parameters.all()]
602 # Remove all conflicting parameters first
603 self.parameters_list.filter(template__pk__in=template_ids).delete()
605 for parameter in other.parameters.all():
606 parameter.pk = None
607 parameter.model_id = self.pk
608 parameter.model_type = content_type
610 parameters.append(parameter)
612 if len(parameters) > 0:
613 common.models.Parameter.objects.bulk_create(parameters, batch_size=250)
615 def get_parameter(self, name: str):
616 """Return a Parameter instance for the given parameter name.
618 Args:
619 name: Name of the parameter template
621 Returns:
622 Parameter instance if found, else None
623 """
624 return self.parameters_list.filter(template__name=name).first()
626 def get_parameters(self) -> QuerySet:
627 """Return all Parameter instances for this model."""
628 return (
629 self.parameters_list
630 .all()
631 .prefetch_related('template', 'model_type')
632 .order_by('template__name')
633 )
635 def parameters_map(self) -> dict:
636 """Return a map (dict) of parameter values associated with this Part instance, of the form.
638 Example:
639 {
640 "name_1": "value_1",
641 "name_2": "value_2",
642 }
643 """
644 params = {}
646 for parameter in self.parameters.all().prefetch_related('template'):
647 params[parameter.template.name] = parameter.data
649 return params
651 def check_parameter_delete(self, parameter):
652 """Run a check to determine if the provided parameter can be deleted.
654 The default implementation always returns True, but this can be overridden in the implementing class.
655 """
656 return True
658 def check_parameter_save(self, parameter):
659 """Run a check to determine if the provided parameter can be saved.
661 The default implementation always returns True, but this can be overridden in the implementing class.
662 """
663 return True
666class InvenTreeAttachmentMixin(InvenTreePermissionCheckMixin):
667 """Provides an abstracted class for managing file attachments.
669 Links the implementing model to the common.models.Attachment table,
670 and provides the following methods:
672 - attachments: Return a queryset containing all attachments for this model
673 """
675 def delete(self, *args, **kwargs):
676 """Handle the deletion of a model instance.
678 Before deleting the model instance, delete any associated attachments.
679 """
680 for attachment in list(self.attachments.all()): 680 ↛ 681line 680 didn't jump to line 681 because the loop on line 680 never started
681 attachment.delete()
683 super().delete(*args, **kwargs)
685 @property
686 def attachments(self) -> QuerySet:
687 """Return a queryset containing all attachments for this model."""
688 return self.attachments_for_model().filter(model_id=self.pk)
690 def attachments_for_model(self) -> QuerySet:
691 """Return all attachments for this model class."""
692 from common.models import Attachment
694 model_type = self.__class__.__name__.lower()
695 return Attachment.objects.filter(model_type=model_type)
697 def create_attachment(self, attachment=None, link=None, comment='', **kwargs):
698 """Create an attachment / link for this model."""
699 from common.models import Attachment
701 kwargs['attachment'] = attachment
702 kwargs['link'] = link
703 kwargs['comment'] = comment
704 kwargs['model_type'] = self.__class__.__name__.lower()
705 kwargs['model_id'] = self.pk
707 Attachment.objects.create(**kwargs)
710class InvenTreeTree(ContentTypeMixin, MPTTModel):
711 """Provides an abstracted self-referencing tree model, based on the MPTTModel class.
713 Our implementation provides the following key improvements:
715 - Allow tracking of separate concepts of "nodes" and "items"
716 - Better handling of deletion of nodes and items
717 - Ensure tree is correctly rebuilt after deletion and other operations
718 - Improved protection against recursive tree structures
719 """
721 # How each node reference its parent object
722 NODE_PARENT_KEY = 'parent'
724 # How items (not nodes) are hooked into the tree
725 # e.g. for StockLocation, this value is 'location'
726 ITEM_PARENT_KEY = None
728 class Meta:
729 """Metaclass defines extra model properties."""
731 abstract = True
733 class MPTTMeta:
734 """MPTT metaclass options."""
736 order_insertion_by = ['name']
738 @transaction.atomic
739 def delete(self, *args, **kwargs):
740 """Handle the deletion of a tree node.
742 kwargs:
743 delete_children: If True, delete all child nodes (otherwise, point to the parent of this node)
744 delete_items: If True, delete all items associated with this node (otherwise, point to the parent of this node)
746 Order of operations:
747 1. Update nodes and items under the current node
748 2. Delete this node
749 3. Rebuild the model tree
750 """
751 delete_children = kwargs.pop('delete_children', False)
752 delete_items = kwargs.pop('delete_items', False)
754 # Ensure that we have the latest version of the database object
755 try:
756 self.refresh_from_db()
757 except self.__class__.DoesNotExist:
758 # If the object no longer exists, raise a ValidationError
759 raise ValidationError(
760 'Object %s of type %s no longer exists', str(self), str(self.__class__)
761 )
763 tree_id = self.tree_id
764 parent = getattr(self, self.NODE_PARENT_KEY, None)
766 # When deleting a top level node with multiple children,
767 # we need to assign a new tree_id to each child node
768 # otherwise they will all have the same tree_id (which is not allowed)
769 lower_trees = []
771 if not parent: # No parent, which means this is a top-level node
772 for child in self.get_children():
773 # Store a flattened list of node IDs for each of the lower trees
774 nodes = list(
775 child
776 .get_descendants(include_self=True)
777 .values_list('pk', flat=True)
778 .distinct()
779 )
780 lower_trees.append(nodes)
782 # 1. Update nodes and items under the current node
783 self.handle_tree_delete(
784 delete_children=delete_children, delete_items=delete_items
785 )
787 # 2. Delete *this* node
788 super().delete(*args, **kwargs)
790 # A set of tree_id values which need to be rebuilt
791 trees = set()
793 if tree_id:
794 # If this node had a tree_id, we need to rebuild that tree
795 trees.add(tree_id)
797 # Did we delete a top-level node?
798 next_tree_id = self.getNextTreeID()
800 # If there is only one sub-tree, it can retain the same tree_id value
801 for tree in lower_trees[1:]:
802 # Bulk update the tree_id for all lower nodes
803 lower_nodes = self.__class__.objects.filter(pk__in=tree)
804 lower_nodes.update(tree_id=next_tree_id)
805 trees.add(next_tree_id)
806 next_tree_id += 1
808 # 3. Rebuild the model tree(s) as required
809 self.__class__.rebuild_trees(trees)
811 def handle_tree_delete(self, delete_children=False, delete_items=False):
812 """Delete a single instance of the tree, based on provided kwargs.
814 Removing a tree "node" from the database must be considered carefully,
815 based on what the user intends for any items which exist *under* that node.
817 - "children" are any nodes (of the same type) which exist *under* this node (e.g. PartCategory)
818 - "items" are any items (of a different type) which exist *under* this node (e.g. Part)
820 Arguments:
821 delete_children: If True, delete all child items
822 delete_items: If True, delete all items associated with this node
824 There are multiple scenarios we can consider here:
826 A) delete_children = True and delete_items = True
827 B) delete_children = True and delete_items = False
828 C) delete_children = False and delete_items = True
829 D) delete_children = False and delete_items = False
830 """
831 child_nodes = self.get_descendants(include_self=False)
833 # Case A: Delete all child items, and all child nodes.
834 # - Delete all items at any lower level
835 # - Delete all descendant nodes
836 if delete_children and delete_items:
837 self.delete_items(cascade=True)
838 self.delete_nodes(child_nodes)
840 # Case B: Delete all child nodes, but move all child items up to the parent
841 # - Move all items at any lower level to the parent of this item
842 # - Delete all descendant nodes
843 elif delete_children and not delete_items:
844 if items := self.get_items(cascade=True):
845 parent = getattr(self, self.NODE_PARENT_KEY, None)
846 items.update(**{self.ITEM_PARENT_KEY: parent})
847 self.delete_nodes(child_nodes)
849 # Case C: Delete all child items, but keep all child nodes
850 # - Remove all items directly associated with this node
851 # - Move any direct child nodes up one level
852 elif not delete_children and delete_items:
853 self.delete_items(cascade=False)
854 parent = getattr(self, self.NODE_PARENT_KEY, None)
855 self.get_children().update(**{self.NODE_PARENT_KEY: parent})
857 # Case D: Keep all child items, and keep all child nodes
858 # - Move all items directly associated with this node up one level
859 # - Move any direct child nodes up one level
860 elif not delete_children and not delete_items:
861 parent = getattr(self, self.NODE_PARENT_KEY, None)
862 if items := self.get_items(cascade=False):
863 items.update(**{self.ITEM_PARENT_KEY: parent})
864 self.get_children().update(**{self.NODE_PARENT_KEY: parent})
866 def delete_nodes(self, nodes):
867 """Delete a set of nodes from the tree.
869 1. First, set the "parent" value for selected nodes to None
870 2. Then, perform bulk deletion of selected nodes
872 Step 1. is required because we cannot guarantee the order-of-operations in the db backend
874 Arguments:
875 nodes: A queryset of nodes to delete
876 """
877 nodes.update(**{self.NODE_PARENT_KEY: None})
878 nodes.delete()
880 def api_instance_filters(self):
881 """Instance filters for InvenTreeTree models."""
882 return {self.NODE_PARENT_KEY: {'exclude_tree': self.pk}}
884 def save(self, *args, **kwargs):
885 """Custom save method for InvenTreeTree abstract model."""
886 db_instance = None
888 parent = getattr(self, self.NODE_PARENT_KEY, None)
890 if not self.tree_id:
891 if parent: 891 ↛ 893line 891 didn't jump to line 893 because the condition on line 891 was never true
892 # If we have a parent, use the parent's tree_id
893 self.tree_id = parent.tree_id
894 self.level = parent.level + 1
895 else:
896 # Otherwise, we need to generate a new tree_id
897 self.tree_id = self.getNextTreeID()
899 if self.pk:
900 try:
901 db_instance = self.get_db_instance()
902 except self.__class__.DoesNotExist:
903 # If the instance does not exist, we cannot get the db instance
904 db_instance = None
905 try:
906 super().save(*args, **kwargs)
907 except InvalidMove:
908 # Provide better error for parent selection
909 raise ValidationError({self.NODE_PARENT_KEY: _('Invalid choice')})
911 trees = set()
913 parent = getattr(self, self.NODE_PARENT_KEY, None)
915 if db_instance:
916 # If the tree_id or parent has changed, we need to rebuild the tree
917 if getattr(db_instance, self.NODE_PARENT_KEY) != parent: 917 ↛ 918line 917 didn't jump to line 918 because the condition on line 917 was never true
918 trees.add(db_instance.tree_id)
919 if db_instance.tree_id != self.tree_id: 919 ↛ 920line 919 didn't jump to line 920 because the condition on line 919 was never true
920 trees.add(self.tree_id)
921 trees.add(db_instance.tree_id)
922 elif parent: 922 ↛ 924line 922 didn't jump to line 924 because the condition on line 922 was never true
923 # New instance, so we need to rebuild the tree (if it has a parent)
924 trees.add(self.tree_id)
926 # Flag to indicate that a tree rebuild task was triggered by this save
927 self._tree_rebuild_offloaded = False
929 if len(trees) > 0: 929 ↛ 933line 929 didn't jump to line 933 because the condition on line 929 was never true
930 # Offload the tree rebuild(s) to the background worker.
931 # Note that repeated calls are de-duplicated (per tree),
932 # so a bulk operation results in a single rebuild per affected tree.
933 ran_sync = self.__class__.offload_tree_rebuild(trees)
934 self._tree_rebuild_offloaded = True
936 if ran_sync:
937 # The tree was rebuilt synchronously, so refresh the instance
938 try:
939 self.refresh_from_db()
940 except TransactionManagementError:
941 # If we are inside a transaction block, we cannot refresh from db
942 pass
943 except Exception as e:
944 # Any other error is unexpected
945 InvenTree.sentry.report_exception(e)
946 InvenTree.exceptions.log_error(f'{self.__class__.__name__}.save')
948 @classmethod
949 def offload_tree_rebuild(cls, tree_ids) -> bool:
950 """Offload a rebuild of the specified trees to the background worker.
952 - The tree structure (and pathstring values, where applicable) are rebuilt for each tree
953 - If the background worker is not running, the rebuild is performed synchronously
954 - Identical pending tasks are skipped, so repeated calls (e.g. during a bulk
955 operation) result in (at most) a single queued rebuild per affected tree
957 Returns:
958 bool: True if any rebuild was performed synchronously (in the calling thread)
959 """
960 from InvenTree.tasks import offload_task
962 ran_sync = False
964 for tree_id in tree_ids:
965 if tree_id:
966 result = offload_task(
967 'InvenTree.tasks.rebuild_model_tree', cls._meta.label_lower, tree_id
968 )
970 if result is True:
971 # offload_task returns True if the task ran synchronously
972 ran_sync = True
974 return ran_sync
976 @classmethod
977 def rebuild_trees(cls, tree_ids) -> None:
978 """Rebuild the specified trees, with fallback to a full rebuild.
980 - Perform a partial rebuild for each provided tree_id
981 - If any partial rebuild fails, rebuild the entire tree (expensive!!!)
982 """
983 result = True
985 for tree_id in tree_ids:
986 if tree_id and not cls.partial_rebuild(tree_id):
987 result = False
989 if not result:
990 # Rebuild the entire tree (expensive!!!)
991 cls.objects.rebuild()
993 @classmethod
994 def partial_rebuild(cls, tree_id: int) -> bool:
995 """Perform a partial rebuild of the tree structure.
997 If a failure occurs, log the error and return False.
998 """
999 try:
1000 cls.objects.partial_rebuild(tree_id)
1001 return True
1002 except Exception as e:
1003 # This is a critical error, explicitly report to sentry
1004 InvenTree.sentry.report_exception(e)
1006 InvenTree.exceptions.log_error(f'{cls.__name__}.partial_rebuild')
1007 logger.exception(
1008 'Failed to rebuild tree <%s> for %s: %s', tree_id, cls.__name__, e
1009 )
1010 return False
1012 def delete_items(self, cascade: bool = False):
1013 """Delete any 'items' which exist under this node in the tree.
1015 - Note that an 'item' is an instance of a different model class.
1016 - Not all tree structures will have items associated with them.
1017 """
1018 if items := self.get_items(cascade=cascade):
1019 items.delete()
1021 def get_items(self, cascade: bool = False):
1022 """Return a queryset of items which exist *under* this node in the tree.
1024 - For a StockLocation instance, this would be a queryset of StockItem objects
1025 - For a PartCategory instance, this would be a queryset of Part objects
1027 The default implementation returns None, indicating that no items exist under this node.
1028 """
1029 return None
1031 def getUniqueParents(self) -> QuerySet:
1032 """Return a flat set of all parent items that exist above this node."""
1033 return self.get_ancestors()
1035 def getUniqueChildren(self, include_self=True) -> QuerySet:
1036 """Return a flat set of all child items that exist under this node."""
1037 return self.get_descendants(include_self=include_self)
1039 @property
1040 def has_children(self) -> bool:
1041 """True if there are any children under this item."""
1042 return self.getUniqueChildren(include_self=False).count() > 0
1044 @classmethod
1045 def getNextTreeID(cls) -> int:
1046 """Return the next available tree_id for this model class."""
1047 instance = cls.objects.order_by('-tree_id').first()
1049 if instance:
1050 return instance.tree_id + 1
1051 else:
1052 return 1
1055class PathStringMixin(models.Model):
1056 """Mixin class for adding a 'pathstring' field to a model class.
1058 The pathstring is a string representation of the path to this model instance,
1059 which can be used for display purposes.
1061 The pathstring is automatically generated when the model instance is saved.
1062 """
1064 # Field to use for constructing a "pathstring" for the tree
1065 PATH_FIELD = 'name'
1067 # Extra fields to include in the get_path result. E.g. icon
1068 EXTRA_PATH_FIELDS = []
1070 class Meta:
1071 """Metaclass options for this mixin.
1073 Note: abstract must be true, as this is only a mixin, not a separate table
1074 """
1076 abstract = True
1078 name = models.CharField(
1079 blank=False, max_length=100, verbose_name=_('Name'), help_text=_('Name')
1080 )
1082 description = models.CharField(
1083 blank=True,
1084 max_length=250,
1085 verbose_name=_('Description'),
1086 help_text=_('Description (optional)'),
1087 )
1089 # When a category is deleted, graft the children onto its parent
1090 parent = TreeForeignKey(
1091 'self',
1092 on_delete=models.DO_NOTHING,
1093 blank=True,
1094 null=True,
1095 verbose_name='parent',
1096 related_name='children',
1097 )
1099 # The 'pathstring' field is calculated each time the model is saved
1100 pathstring = models.CharField(
1101 blank=True, max_length=250, verbose_name=_('Path'), help_text=_('Path')
1102 )
1104 def save(self, *args, **kwargs):
1105 """Update the pathstring field when saving the model instance."""
1106 old_pathstring = self.pathstring
1108 # Rebuild upper first, to ensure the lower nodes are updated correctly
1109 super().save(*args, **kwargs)
1111 # Determine if a tree rebuild task was already triggered by this save
1112 # (e.g. if the node was re-parented) - if so, the pathstring values
1113 # for any lower nodes are updated by that task
1114 rebuild_offloaded = getattr(self, '_tree_rebuild_offloaded', False)
1116 # Ensure that the pathstring is correctly constructed
1117 pathstring = self.construct_pathstring(refresh=True)
1119 if pathstring != old_pathstring:
1120 kwargs.pop('force_insert', None)
1121 kwargs['force_update'] = True
1123 self.pathstring = pathstring
1124 super().save(*args, **kwargs)
1126 # Update the pathstring values for any lower nodes,
1127 # by offloading the update to the background worker
1128 if not rebuild_offloaded and self.get_descendant_count() > 0: 1128 ↛ 1129line 1128 didn't jump to line 1129 because the condition on line 1128 was never true
1129 self.__class__.offload_tree_rebuild([self.tree_id])
1131 def delete(self, *args, **kwargs):
1132 """Custom delete method for PathStringMixin.
1134 - Before deleting the object, update the pathstring for any child nodes.
1135 - Then, delete the object.
1136 """
1137 # Ensure that we have the latest version of the database object
1138 try:
1139 self.refresh_from_db()
1140 except self.__class__.DoesNotExist:
1141 # If the object no longer exists, raise a ValidationError
1142 raise ValidationError(
1143 'Object %s of type %s no longer exists', str(self), str(self.__class__)
1144 )
1146 # Store the node ID values for lower nodes, before we delete this one
1147 lower_nodes = self.get_lower_nodes()
1149 # Delete this node - after which we expect the tree structure will be updated
1150 super().delete(*args, **kwargs)
1152 # Rebuild the pathstring for lower nodes
1153 self.rebuild_lower_nodes(lower_nodes)
1155 def __str__(self):
1156 """String representation of a category is the full path to that category."""
1157 return f'{self.pathstring} - {self.description}'
1159 def get_lower_nodes(self) -> list[int]:
1160 """Return a list of all lower nodes in the tree."""
1161 return list(
1162 self.get_descendants(include_self=False).values_list('pk', flat=True)
1163 )
1165 def rebuild_lower_nodes(self, lower_nodes: list[int]):
1166 """Rebuild the pathstring for lower nodes in the tree.
1168 - This is used when the pathstring for this node is updated, and we need to update all lower nodes.
1169 - We use a bulk-update to update the pathstring for all lower nodes in the tree.
1170 """
1171 nodes = self.__class__.objects.filter(pk__in=lower_nodes)
1173 nodes_to_update = []
1175 for node in nodes:
1176 new_path = node.construct_pathstring()
1178 if new_path != node.pathstring:
1179 node.pathstring = new_path
1180 nodes_to_update.append(node)
1182 if len(nodes_to_update) > 0:
1183 self.__class__.objects.bulk_update(nodes_to_update, ['pathstring'])
1185 @classmethod
1186 def rebuild_tree_pathstring_values(cls, tree_ids) -> None:
1187 """Rebuild the 'pathstring' values for all nodes in the specified trees.
1189 Each tree is processed in a single pass:
1190 the pathstring for each node is constructed from its parent node,
1191 and any changed values are written back in a single bulk update.
1192 """
1193 tree_nodes = list(cls.objects.filter(tree_id__in=tree_ids))
1194 node_map = {node.pk: node for node in tree_nodes}
1196 # Cache of node ID -> list of path elements (from the top level down)
1197 path_cache: dict[int, list[str]] = {}
1199 def path_names(node) -> list[str]:
1200 """Construct the path (list of names) for a node, via its parent chain."""
1201 if node.pk in path_cache:
1202 return path_cache[node.pk]
1204 names = [str(getattr(node, cls.PATH_FIELD, node.pk))]
1206 if node.parent_id:
1207 parent = node_map.get(node.parent_id)
1209 if parent is None:
1210 # Parent node exists outside the selected trees
1211 parent = cls.objects.get(pk=node.parent_id)
1212 node_map[node.parent_id] = parent
1214 names = [*path_names(parent), *names]
1216 path_cache[node.pk] = names
1217 return names
1219 nodes_to_update = []
1221 for node in tree_nodes:
1222 pathstring = InvenTree.helpers.constructPathString(path_names(node))
1224 if pathstring != node.pathstring:
1225 node.pathstring = pathstring
1226 nodes_to_update.append(node)
1228 if len(nodes_to_update) > 0:
1229 cls.objects.bulk_update(nodes_to_update, ['pathstring'], batch_size=250)
1231 def construct_pathstring(self, refresh: bool = False) -> str:
1232 """Construct the pathstring for this tree node.
1234 Arguments:
1235 refresh: If True, force a refresh of the model instance
1236 """
1237 if refresh: 1237 ↛ 1241line 1237 didn't jump to line 1241 because the condition on line 1237 was always true
1238 # Refresh the model instance from the database
1239 self.refresh_from_db()
1241 return InvenTree.helpers.constructPathString([
1242 getattr(item, self.PATH_FIELD, item.pk) for item in self.path
1243 ])
1245 def validate_unique(self, exclude=None):
1246 """Validate that this tree instance satisfies our uniqueness requirements.
1248 Note that a 'unique_together' requirement for ('name', 'parent') is insufficient,
1249 as it ignores cases where parent=None (i.e. top-level items)
1250 """
1251 super().validate_unique(exclude)
1253 results = self.__class__.objects.filter(
1254 name=self.name, parent=self.parent
1255 ).exclude(pk=self.pk)
1257 if results.exists():
1258 raise ValidationError(
1259 _('Duplicate names cannot exist under the same parent')
1260 )
1262 @property
1263 def parentpath(self) -> list:
1264 """Get the parent path of this category.
1266 Returns:
1267 List of category names from the top level to the parent of this category
1268 """
1269 return list(self.get_ancestors())
1271 @property
1272 def path(self) -> list:
1273 """Get the complete part of this category.
1275 e.g. ["Top", "Second", "Third", "This"]
1277 Returns:
1278 List of category names from the top level to this category
1279 """
1280 return [*self.parentpath, self]
1282 def get_path(self) -> list:
1283 """Return a list of element in the item tree.
1285 Contains the full path to this item, with each entry containing the following data:
1287 {
1288 pk: <pk>,
1289 name: <name>,
1290 }
1291 """
1292 return [
1293 {
1294 'pk': item.pk,
1295 'name': getattr(item, self.PATH_FIELD, item.pk),
1296 **{k: getattr(item, k, None) for k in self.EXTRA_PATH_FIELDS},
1297 }
1298 for item in self.path
1299 ]
1302class InvenTreeNotesMixin(models.Model):
1303 """A mixin class for adding notes functionality to a model class.
1305 The following fields are added to any model which implements this mixin:
1307 - notes : A text field for storing notes
1308 """
1310 class Meta:
1311 """Metaclass options for this mixin.
1313 Note: abstract must be true, as this is only a mixin, not a separate table
1314 """
1316 abstract = True
1318 def delete(self, *args, **kwargs):
1319 """Custom delete method for InvenTreeNotesMixin.
1321 - Before deleting the object, check if there are any uploaded images associated with it.
1322 - If so, delete the notes first
1323 """
1324 from common.models import NotesImage
1326 images = NotesImage.objects.filter(
1327 model_type=self.__class__.__name__.lower(), model_id=self.pk
1328 )
1330 if images.exists(): 1330 ↛ 1331line 1330 didn't jump to line 1331 because the condition on line 1330 was never true
1331 logger.info(
1332 'Deleting %s uploaded images associated with %s <%s>',
1333 images.count(),
1334 self.__class__.__name__,
1335 self.pk,
1336 )
1338 images.delete()
1340 super().delete(*args, **kwargs)
1342 notes = InvenTree.fields.InvenTreeNotesField(
1343 verbose_name=_('Notes'), help_text=_('Markdown notes (optional)')
1344 )
1347class InvenTreeTagsMixin(models.Model):
1348 """A mixin class for adding tag functionality to a model class.
1350 The following fields are added to any model which implements this mixin:
1352 - tags : A text field for storing comma-separated tags
1353 """
1355 class Meta:
1356 """Metaclass options for this mixin.
1358 Note: abstract must be true, as this is only a mixin, not a separate table
1359 """
1361 abstract = True
1363 tags = TaggableManager(blank=True)
1366class InvenTreeBarcodeMixin(models.Model):
1367 """A mixin class for adding barcode functionality to a model class.
1369 Two types of barcodes are supported:
1371 - Internal barcodes (QR codes using a strictly defined format)
1372 - External barcodes (assign third party barcode data to a model instance)
1374 The following fields are added to any model which implements this mixin:
1376 - barcode_data : Raw data associated with an assigned barcode
1377 - barcode_hash : A 'hash' of the assigned barcode data used to improve matching
1379 The barcode_model_type_code() classmethod must be implemented in the model class.
1380 """
1382 class Meta:
1383 """Metaclass options for this mixin.
1385 Note: abstract must be true, as this is only a mixin, not a separate table
1386 """
1388 abstract = True
1390 barcode_data = models.CharField(
1391 blank=True,
1392 max_length=500,
1393 verbose_name=_('Barcode Data'),
1394 help_text=_('Third party barcode data'),
1395 )
1397 barcode_hash = models.CharField(
1398 blank=True,
1399 max_length=128,
1400 verbose_name=_('Barcode Hash'),
1401 help_text=_('Unique hash of barcode data'),
1402 )
1404 @classmethod
1405 def barcode_model_type(cls):
1406 """Return the model 'type' for creating a custom QR code."""
1407 # By default, use the name of the class
1408 return cls.__name__.lower()
1410 @classmethod
1411 def barcode_model_type_code(cls):
1412 r"""Return a 'short' code for the model type.
1414 This is used to generate a efficient QR code for the model type.
1415 It is expected to match this pattern: [0-9A-Z $%*+-.\/:]{2}
1417 Note: Due to the shape constraints (45**2=2025 different allowed codes)
1418 this needs to be explicitly implemented in the model class to avoid collisions.
1419 """
1420 raise NotImplementedError(
1421 'barcode_model_type_code() must be implemented in the model class'
1422 )
1424 def format_barcode(self, **kwargs):
1425 """Return a string for formatting a QR code for this model instance."""
1426 from plugin.base.barcodes.helper import generate_barcode
1428 return generate_barcode(self)
1430 def format_matched_response(self, user, **kwargs):
1431 """Format a standard response for a matched barcode."""
1432 # Check permission for this object
1433 from users.permissions import check_user_permission
1435 if not check_user_permission(user, self, 'view'):
1436 raise PermissionDenied(
1437 _('User does not have permission to view this model')
1438 )
1440 data = {'pk': self.pk}
1442 if hasattr(self, 'get_api_url'):
1443 api_url = self.get_api_url()
1444 data['api_url'] = api_url = f'{api_url}{self.pk}/'
1446 # Attempt to serialize the object too
1447 try:
1448 match = resolve(api_url)
1449 view_class = match.func.view_class
1450 serializer_class = view_class.serializer_class
1451 serializer = serializer_class(self)
1452 data['instance'] = serializer.data
1453 except Exception:
1454 pass
1456 if hasattr(self, 'get_absolute_url'):
1457 data['web_url'] = self.get_absolute_url()
1459 return data
1461 @property
1462 def barcode(self) -> str:
1463 """Format a minimal barcode string (e.g. for label printing)."""
1464 return self.format_barcode()
1466 @classmethod
1467 def lookup_barcode(cls, barcode_hash: str) -> models.Model:
1468 """Check if a model instance exists with the specified third-party barcode hash."""
1469 return cls.objects.filter(barcode_hash=barcode_hash).first()
1471 def assign_barcode(
1472 self,
1473 barcode_hash: Optional[str] = None,
1474 barcode_data: Optional[str] = None,
1475 raise_error: bool = True,
1476 save: bool = True,
1477 ):
1478 """Assign an external (third-party) barcode to this object."""
1479 # Must provide either barcode_hash or barcode_data
1480 if barcode_hash is None and barcode_data is None:
1481 raise ValueError("Provide either 'barcode_hash' or 'barcode_data'")
1483 # If barcode_hash is not provided, create from supplier barcode_data
1484 if barcode_hash is None and barcode_data is not None:
1485 barcode_hash = InvenTree.helpers.hash_barcode(barcode_data)
1487 # Check for existing item
1488 if self.__class__.lookup_barcode(barcode_hash) is not None:
1489 if raise_error:
1490 raise ValidationError(_('Existing barcode found'))
1491 else:
1492 return False
1494 if barcode_data is not None:
1495 self.barcode_data = barcode_data
1497 self.barcode_hash = barcode_hash
1499 if save:
1500 self.save()
1502 return True
1504 def unassign_barcode(self):
1505 """Unassign custom barcode from this model."""
1506 self.barcode_data = ''
1507 self.barcode_hash = ''
1509 self.save()
1512def notify_staff_users_of_error(instance, label: str, context: dict):
1513 """Helper function to notify staff users of an error."""
1514 import common.models
1515 import common.notifications
1516 from plugin.builtin.integration.core_notifications import InvenTreeUINotifications
1518 try:
1519 # Get all staff users
1520 staff_users = get_user_model().objects.filter(is_active=True, is_staff=True)
1522 target_users = []
1524 # Send a notification to each staff user (unless they have disabled error notifications)
1525 for user in staff_users:
1526 if common.models.InvenTreeUserSetting.get_setting( 1526 ↛ 1525line 1526 didn't jump to line 1525 because the condition on line 1526 was always true
1527 'NOTIFICATION_ERROR_REPORT', True, user=user
1528 ):
1529 target_users.append(user)
1531 if len(target_users) > 0: 1531 ↛ exitline 1531 didn't return from function 'notify_staff_users_of_error' because the condition on line 1531 was always true
1532 common.notifications.trigger_notification(
1533 instance,
1534 label,
1535 context=context,
1536 targets=target_users,
1537 delivery_methods={InvenTreeUINotifications},
1538 )
1540 except Exception as exc:
1541 # We do not want to throw an exception while reporting an exception!
1542 logger.error(exc)
1545@receiver(post_save, sender=Task, dispatch_uid='failure_post_save_notification')
1546def after_failed_task(sender, instance: Task, created: bool, **kwargs):
1547 """Callback when a new task failure log is generated."""
1548 from django.conf import settings
1550 from InvenTree.exceptions import log_error
1552 max_attempts = int(settings.Q_CLUSTER.get('max_attempts', 5))
1553 n = instance.attempt_count
1555 # Only notify once the maximum number of attempts has been reached
1556 if not instance.success and n >= max_attempts:
1557 # Create a new Error object associated with this failed task
1558 # This will, in turn, trigger a notification to staff users via the Error post_save signal
1560 log_error(
1561 'task_failure',
1562 scope='worker',
1563 error_name='Task Failure',
1564 error_info=f"Task '{instance.pk}' failed after {n} attempts",
1565 error_data=str(instance.result) if instance.result else '',
1566 )
1569@receiver(post_save, sender=Error, dispatch_uid='error_post_save_notification')
1570def after_error_logged(sender, instance: Error, created: bool, **kwargs):
1571 """Callback when a server error is logged.
1573 - Send a UI notification to all users with staff status
1574 """
1575 if created: 1575 ↛ exitline 1575 didn't return from function 'after_error_logged' because the condition on line 1575 was always true
1576 try:
1577 url = InvenTree.helpers_model.construct_absolute_url(
1578 reverse(
1579 'admin:error_report_error_change', kwargs={'object_id': instance.pk}
1580 )
1581 )
1582 except NoReverseMatch:
1583 url = ''
1585 notify_staff_users_of_error(
1586 instance,
1587 'inventree.error_log',
1588 {
1589 'error': instance,
1590 'name': _('Server Error'),
1591 'message': _('An error has been logged by the server.'),
1592 'link': url,
1593 },
1594 )
1597class InvenTreeImageMixin(models.Model):
1598 """A mixin class for adding image functionality to a model class.
1600 The following fields are added to any model which implements this mixin:
1602 - image : An image field for storing an image
1603 """
1605 IMAGE_RENAME: Callable | None = None
1607 class Meta:
1608 """Metaclass options for this mixin.
1610 Note: abstract must be true, as this is only a mixin, not a separate table
1611 """
1613 abstract = True
1615 def __init__(self, *args: Any, **kwargs: Any) -> None:
1616 """Custom init method for InvenTreeImageMixin to ensure IMAGE_RENAME is implemented."""
1617 if self.IMAGE_RENAME is None: 1617 ↛ 1618line 1617 didn't jump to line 1618 because the condition on line 1617 was never true
1618 raise NotImplementedError(
1619 'IMAGE_RENAME must be implemented in the model class'
1620 )
1621 super().__init__(*args, **kwargs)
1623 def rename_image(self, filename):
1624 """Rename the uploaded image file using the IMAGE_RENAME function."""
1625 return self.IMAGE_RENAME(filename)
1627 image = StdImageField(
1628 upload_to=rename_image,
1629 null=True,
1630 blank=True,
1631 variations={'thumbnail': (128, 128), 'preview': (256, 256)},
1632 delete_orphans=False,
1633 verbose_name=_('Image'),
1634 )
1636 def get_image_url(self):
1637 """Return the URL of the image for this object."""
1638 if self.image:
1639 return InvenTree.helpers.getMediaUrl(self.image)
1640 return InvenTree.helpers.getBlankImage()
1642 def get_thumbnail_url(self) -> str:
1643 """Return the URL of the image thumbnail for this object."""
1644 if self.image: 1644 ↛ 1645line 1644 didn't jump to line 1645 because the condition on line 1644 was never true
1645 return InvenTree.helpers.getMediaUrl(self.image, 'thumbnail')
1646 return InvenTree.helpers.getBlankThumbnail()