Coverage for src/backend/InvenTree/common/models.py: 56%
1345 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 17:47 +0000
1"""Common database model definitions.
3These models are 'generic' and do not fit a particular business logic object.
4"""
6import base64
7import hashlib
8import hmac
9import json
10import math
11import os
12import uuid
13from collections import OrderedDict
14from datetime import timedelta, timezone
15from email.utils import make_msgid
16from enum import Enum
17from io import BytesIO
18from pathlib import Path
19from secrets import compare_digest
20from typing import Any, Optional
22from django.apps import apps
23from django.conf import settings as django_settings
24from django.contrib.auth.models import User
25from django.contrib.contenttypes.fields import GenericForeignKey
26from django.contrib.contenttypes.models import ContentType
27from django.contrib.humanize.templatetags.humanize import naturaltime
28from django.core.cache import cache
29from django.core.exceptions import SuspiciousFileOperation, ValidationError
30from django.core.files.base import ContentFile
31from django.core.files.storage import default_storage
32from django.core.files.utils import validate_file_name
33from django.core.mail import EmailMultiAlternatives, get_connection
34from django.core.mail.utils import DNS_NAME
35from django.core.validators import MinLengthValidator, MinValueValidator
36from django.db import models, transaction
37from django.db.models import enums
38from django.db.models.signals import post_delete, post_save
39from django.db.utils import IntegrityError, OperationalError, ProgrammingError
40from django.dispatch import receiver
41from django.urls import reverse
42from django.utils.timezone import now
43from django.utils.translation import gettext_lazy as _
45import structlog
46from anymail.signals import inbound, tracking
47from django_q.signals import post_spawn
48from djmoney.contrib.exchange.exceptions import MissingRate
49from djmoney.contrib.exchange.models import convert_money
50from opentelemetry import trace
51from PIL import Image
52from rest_framework.exceptions import PermissionDenied
54import common.validators
55import InvenTree.conversion
56import InvenTree.exceptions
57import InvenTree.fields
58import InvenTree.helpers
59import InvenTree.models
60import InvenTree.ready
61import InvenTree.tasks
62import InvenTree.validators
63import users.models
64from common.setting.type import InvenTreeSettingsKeyType, SettingsKeyType
65from common.settings import get_global_setting, global_setting_overrides
66from generic.enums import StringEnum
67from generic.states import ColorEnum
68from generic.states.custom import state_color_mappings
69from InvenTree.cache import get_session_cache, set_session_cache
70from InvenTree.sanitizer import sanitize_svg
71from InvenTree.tracing import TRACE_PROC, TRACE_PROV
72from InvenTree.version import inventree_identifier
74logger = structlog.get_logger('inventree')
77class RenderMeta(enums.ChoicesType):
78 """Metaclass for rendering choices."""
80 choice_fnc = None
82 @property
83 def choices(self):
84 """Return a list of choices for the enum class."""
85 fnc = getattr(self, 'choice_fnc', None)
86 if fnc: 86 ↛ 88line 86 didn't jump to line 88 because the condition on line 86 was always true
87 return fnc()
88 return []
91class RenderChoices(models.TextChoices, metaclass=RenderMeta):
92 """Class for creating enumerated string choices for schema rendering.""" # ty:ignore[conflicting-metaclass]
95class MetaMixin(models.Model):
96 """A base class for InvenTree models to include shared meta fields.
98 Attributes:
99 - updated: The last time this object was updated
100 """
102 class Meta:
103 """Meta options for MetaMixin."""
105 abstract = True
107 updated = models.DateTimeField(
108 verbose_name=_('Updated'),
109 help_text=_('Timestamp of last update'),
110 auto_now=True,
111 null=True,
112 )
115class UpdatedUserMixin(models.Model):
116 """A mixin which stores additional information about the user who created or last modified the object."""
118 class Meta:
119 """Meta options for MetaUserMixin."""
121 abstract = True
123 def save(self, *args, **kwargs):
124 """Extract the user object from kwargs, if provided."""
125 if updated_by := kwargs.pop('updated_by', None): 125 ↛ 126line 125 didn't jump to line 126 because the condition on line 125 was never true
126 self.updated_by = updated_by
128 self.updated = InvenTree.helpers.current_time()
130 super().save(*args, **kwargs)
132 updated = models.DateTimeField(
133 verbose_name=_('Updated'),
134 help_text=_('Timestamp of last update'),
135 default=None,
136 blank=True,
137 null=True,
138 )
140 updated_by = models.ForeignKey(
141 User,
142 on_delete=models.SET_NULL,
143 null=True,
144 blank=True,
145 related_name='%(class)s_updated',
146 verbose_name=_('Update By'),
147 help_text=_('User who last updated this object'),
148 )
151class ProjectCode(InvenTree.models.InvenTreeMetadataModel):
152 """A ProjectCode is a unique identifier for a project."""
154 IMPORT_ID_FIELDS = ['code']
156 class Meta:
157 """Class options for the ProjectCode model."""
159 verbose_name = _('Project Code')
161 @staticmethod
162 def get_api_url():
163 """Return the API URL for this model."""
164 return reverse('api-project-code-list')
166 def __str__(self):
167 """String representation of a ProjectCode."""
168 return self.code
170 code = models.CharField(
171 max_length=50,
172 unique=True,
173 verbose_name=_('Project Code'),
174 help_text=_('Unique project code'),
175 )
177 description = models.CharField(
178 max_length=200,
179 blank=True,
180 verbose_name=_('Description'),
181 help_text=_('Project description'),
182 )
184 responsible = models.ForeignKey(
185 users.models.Owner,
186 on_delete=models.SET_NULL,
187 blank=True,
188 null=True,
189 verbose_name=_('Responsible'),
190 help_text=_('User or group responsible for this project'),
191 related_name='project_codes',
192 )
195class BaseInvenTreeSetting(models.Model):
196 """An base InvenTreeSetting object is a key:value pair used for storing single values (e.g. one-off settings values).
198 Attributes:
199 SETTINGS: definition of all available settings
200 extra_unique_fields: List of extra fields used to be unique, e.g. for PluginConfig -> plugin
201 """
203 SETTINGS: dict[str, SettingsKeyType] = {}
205 CHECK_SETTING_KEY = False
207 extra_unique_fields: list[str] = []
209 class Meta:
210 """Meta options for BaseInvenTreeSetting -> abstract stops creation of database entry."""
212 abstract = True
214 def save(self, *args, **kwargs):
215 """Enforce validation and clean before saving."""
216 self.key = str(self.key).upper()
218 do_cache = kwargs.pop('cache', True)
220 self.clean()
221 self.validate_unique()
223 # Execute before_save action
224 self._call_settings_function('before_save', args, kwargs)
226 super().save()
228 # Update this setting in the cache after it was saved so a pk exists
229 if do_cache: 229 ↛ 233line 229 didn't jump to line 233 because the condition on line 229 was always true
230 self.save_to_cache()
232 # Remove the setting from the request cache
233 set_session_cache(self.cache_key, None)
235 # Execute after_save action
236 self._call_settings_function('after_save', args, kwargs)
238 @classmethod
239 def build_default_values(cls, **kwargs):
240 """Ensure that all values defined in SETTINGS are present in the database.
242 If a particular setting is not present, create it with the default value
243 """
244 try:
245 existing_keys = cls.objects.filter(**kwargs).values_list('key', flat=True)
246 settings_keys = cls.SETTINGS.keys()
248 missing_keys = set(settings_keys) - set(existing_keys)
250 if len(missing_keys) > 0:
251 logger.info('Building %s default values for %s', len(missing_keys), cls)
252 cls.objects.bulk_create(
253 [
254 cls(key=key, value=cls.get_setting_default(key), **kwargs)
255 for key in missing_keys
256 if not key.startswith('_')
257 ],
258 batch_size=250,
259 )
260 except Exception as exc:
261 logger.exception(
262 'Failed to build default values for %s (%s)', cls, type(exc)
263 )
265 def _call_settings_function(self, reference: str, args, kwargs):
266 """Call a function associated with a particular setting.
268 Args:
269 reference (str): The name of the function to call
270 args: Positional arguments to pass to the function
271 kwargs: Keyword arguments to pass to the function
272 """
273 # Get action
274 setting = self.get_setting_definition(
275 self.key, *args, **{**self.get_filters_for_instance(), **kwargs}
276 )
277 settings_fnc = setting.get(reference, None)
279 # Execute if callable
280 if callable(settings_fnc):
281 settings_fnc(self)
283 @property
284 def cache_key(self):
285 """Generate a unique cache key for this settings object."""
286 return self.__class__.create_cache_key(
287 self.key, **self.get_filters_for_instance()
288 )
290 def save_to_cache(self):
291 """Save this setting object to cache."""
292 key = self.cache_key
294 # skip saving to cache if no pk is set
295 if self.pk is None:
296 return
298 logger.debug("Saving setting '%s' to cache", key)
300 try:
301 cache.set(key, self, timeout=3600)
302 except Exception: # pragma: no cover
303 pass
305 @classmethod
306 def create_cache_key(cls, setting_key, **kwargs):
307 """Create a unique cache key for a particular setting object.
309 The cache key uses the following elements to ensure the key is 'unique':
310 - The name of the class
311 - The unique KEY string
312 - Any key:value kwargs associated with the particular setting type (e.g. user-id)
313 """
314 key = f'{cls.__name__!s}:{setting_key}'
316 for k, v in kwargs.items():
317 key += f'_{k}:{v}'
319 return key.replace(' ', '')
321 @classmethod
322 def get_filters(cls, **kwargs):
323 """Enable to filter by other kwargs defined in cls.extra_unique_fields."""
324 return {
325 key: value
326 for key, value in kwargs.items()
327 if key in cls.extra_unique_fields
328 }
330 def get_filters_for_instance(self):
331 """Enable to filter by other fields defined in self.extra_unique_fields."""
332 return {
333 key: getattr(self, key, None)
334 for key in self.extra_unique_fields
335 if hasattr(self, key)
336 }
338 @classmethod
339 def all_settings(
340 cls,
341 *,
342 exclude_hidden=False,
343 settings_definition: dict[str, SettingsKeyType] | None = None,
344 **kwargs,
345 ):
346 """Return a list of "all" defined settings.
348 This performs a single database lookup,
349 and then any settings which are not *in* the database
350 are assigned their default values
351 """
352 filters = cls.get_filters(**kwargs)
354 results = cls.objects.all()
356 if exclude_hidden: 356 ↛ 358line 356 didn't jump to line 358 because the condition on line 356 was never true
357 # Keys which start with an underscore are used for internal functionality
358 results = results.exclude(key__startswith='_')
360 # Optionally filter by other keys
361 results = results.filter(**filters)
363 settings: dict[str, BaseInvenTreeSetting] = {}
365 # Query the database
366 for setting in results:
367 if setting.key: 367 ↛ 366line 367 didn't jump to line 366 because the condition on line 367 was always true
368 settings[setting.key.upper()] = setting
370 # Specify any "default" values which are not in the database
371 settings_definition = settings_definition or cls.SETTINGS
373 all_settings = OrderedDict()
375 for key, setting in settings_definition.items():
376 # If the setting is already in the database, use that value
377 if key.upper() in settings: 377 ↛ 380line 377 didn't jump to line 380 because the condition on line 377 was always true
378 all_settings[key] = settings[key.upper()]
379 else:
380 all_settings[key.upper()] = cls(
381 key=key.upper(),
382 value=cls.get_setting_default(key, **filters),
383 **filters,
384 )
386 # remove any hidden settings
387 if exclude_hidden and setting.get('hidden', False): 387 ↛ 388line 387 didn't jump to line 388 because the condition on line 387 was never true
388 del all_settings[key.upper()]
390 # format settings values and remove protected
391 for key, setting in all_settings.items():
392 validator = cls.get_setting_validator(key, **filters)
394 if cls.is_protected(key, **filters) and setting.value != '': 394 ↛ 395line 394 didn't jump to line 395 because the condition on line 394 was never true
395 setting.value = '***'
396 elif cls.validator_is_bool(validator): 396 ↛ 397line 396 didn't jump to line 397 because the condition on line 396 was never true
397 setting.value = InvenTree.helpers.str2bool(setting.value)
398 elif cls.validator_is_int(validator): 398 ↛ 399line 398 didn't jump to line 399 because the condition on line 398 was never true
399 try:
400 setting.value = int(setting.value)
401 except ValueError:
402 setting.value = cls.get_setting_default(key, **filters)
404 return all_settings
406 @classmethod
407 def allValues(
408 cls,
409 *,
410 exclude_hidden=False,
411 settings_definition: dict[str, SettingsKeyType] | None = None,
412 **kwargs,
413 ):
414 """Return a dict of "all" defined global settings.
416 This performs a single database lookup,
417 and then any settings which are not *in* the database
418 are assigned their default values
419 """
420 all_settings = cls.all_settings(
421 exclude_hidden=exclude_hidden,
422 settings_definition=settings_definition,
423 **kwargs,
424 )
426 settings: dict[str, Any] = {}
428 for key, setting in all_settings.items():
429 settings[key] = setting.value
431 return settings
433 @classmethod
434 def check_all_settings(
435 cls,
436 *,
437 exclude_hidden=False,
438 settings_definition: dict[str, SettingsKeyType] | None = None,
439 **kwargs,
440 ):
441 """Check if all required settings are set by definition.
443 Returns:
444 is_valid: Are all required settings defined
445 missing_settings: List of all settings that are missing (empty if is_valid is 'True')
446 """
447 all_settings = cls.all_settings(
448 exclude_hidden=exclude_hidden,
449 settings_definition=settings_definition,
450 **kwargs,
451 )
453 missing_settings: list[str] = []
455 for setting in all_settings.values():
456 if setting.required:
457 value = setting.value or cls.get_setting_default(setting.key, **kwargs)
459 if value == '':
460 missing_settings.append(setting.key.upper())
462 return len(missing_settings) == 0, missing_settings
464 @classmethod
465 def get_setting_definition(cls, key, **kwargs):
466 """Return the 'definition' of a particular settings value, as a dict object.
468 - The 'settings' dict can be passed as a kwarg
469 - If not passed, look for cls.SETTINGS
470 - Returns an empty dict if the key is not found
471 """
472 settings = kwargs.get('settings', cls.SETTINGS)
474 key = str(key).strip().upper()
476 if settings is not None and key in settings:
477 return settings[key]
478 return {}
480 @classmethod
481 def get_setting_name(cls, key, **kwargs):
482 """Return the name of a particular setting.
484 If it does not exist, return an empty string.
485 """
486 setting = cls.get_setting_definition(key, **kwargs)
487 return setting.get('name', '')
489 @classmethod
490 def get_setting_description(cls, key, **kwargs):
491 """Return the description for a particular setting.
493 If it does not exist, return an empty string.
494 """
495 setting = cls.get_setting_definition(key, **kwargs)
497 return setting.get('description', '')
499 @classmethod
500 def get_setting_units(cls, key, **kwargs):
501 """Return the units for a particular setting.
503 If it does not exist, return an empty string.
504 """
505 setting = cls.get_setting_definition(key, **kwargs)
507 return setting.get('units', '')
509 @classmethod
510 def get_setting_validator(cls, key, **kwargs):
511 """Return the validator for a particular setting.
513 If it does not exist, return None
514 """
515 setting = cls.get_setting_definition(key, **kwargs)
517 return setting.get('validator', None)
519 @classmethod
520 def get_setting_default(cls, key, **kwargs):
521 """Return the default value for a particular setting.
523 If it does not exist, return an empty string
524 """
525 setting = cls.get_setting_definition(key, **kwargs)
527 default = setting.get('default', '')
529 if callable(default):
530 return default()
531 return default
533 @classmethod
534 def get_setting_choices(cls, key, **kwargs):
535 """Return the validator choices available for a particular setting."""
536 setting = cls.get_setting_definition(key, **kwargs)
538 choices = setting.get('choices', None)
540 if callable(choices):
541 # Evaluate the function (we expect it will return a list of tuples...)
542 try:
543 # Attempt to pass the kwargs to the function, if it doesn't expect them, ignore and call without
544 return choices(**kwargs)
545 except TypeError:
546 return choices()
548 return choices
550 @classmethod
551 def get_setting_object(cls, key, **kwargs):
552 """Return an InvenTreeSetting object matching the given key.
554 - Key is case-insensitive
555 - Returns None if no match is made
557 As settings are accessed frequently, this function will attempt to access the cache first:
559 1. Check the ephemeral request cache
560 2. Check the global cache
561 3. Query the database
562 """
563 key = str(key).strip().upper()
565 # Unless otherwise specified, attempt to create the setting
566 create = kwargs.pop('create', True)
568 # Specify if global cache lookup should be performed
569 # If not specified, determine based on whether global cache is enabled
570 access_global_cache = kwargs.pop('cache', django_settings.GLOBAL_CACHE_ENABLED)
572 # Prevent saving to the database during certain operations
573 if ( 573 ↛ 579line 573 didn't jump to line 579 because the condition on line 573 was never true
574 InvenTree.ready.isImportingData()
575 or InvenTree.ready.isRunningMigrations()
576 or InvenTree.ready.isRebuildingData()
577 or InvenTree.ready.isRunningBackup()
578 ): # pragma: no cover
579 create = False
580 access_global_cache = False
582 cache_key = cls.create_cache_key(key, **kwargs)
584 # Fist, attempt to pull the setting from the request cache
585 if setting := get_session_cache(cache_key):
586 return setting
588 if access_global_cache:
589 try:
590 # First attempt to find the setting object in the cache
591 cached_setting = cache.get(cache_key)
593 if cached_setting is not None:
594 # Store the cached setting into the session cache
596 set_session_cache(cache_key, cached_setting)
597 return cached_setting
599 except Exception:
600 # Cache is not ready yet
601 access_global_cache = False
603 # At this point, we need to query the database
605 filters = {
606 'key__iexact': key,
607 # Optionally filter by other keys
608 **cls.get_filters(**kwargs),
609 }
611 try:
612 settings = cls.objects.all()
613 setting = settings.filter(**filters).first()
614 except (ValueError, cls.DoesNotExist):
615 setting = None
616 except (IntegrityError, OperationalError, ProgrammingError):
617 setting = None
619 # Setting does not exist! (Try to create it)
620 if not setting and create:
621 # Attempt to create a new settings object
622 default_value = cls.get_setting_default(key, **kwargs)
624 extra_fields = {}
626 # Provide extra default fields
627 for field in cls.extra_unique_fields:
628 if field in kwargs: 628 ↛ 627line 628 didn't jump to line 627 because the condition on line 628 was always true
629 extra_fields[field] = kwargs[field]
631 setting = cls(key=key, value=default_value, **extra_fields)
633 try:
634 # Wrap this statement in "atomic", so it can be rolled back if it fails
635 with transaction.atomic():
636 setting.save(**kwargs)
637 except (IntegrityError, OperationalError, ProgrammingError):
638 # It might be the case that the database isn't created yet
639 pass
640 except ValidationError:
641 # The setting failed validation - might be due to duplicate keys
642 pass
644 if setting:
645 # Cache this setting object to the request cache
646 set_session_cache(cache_key, setting)
648 if access_global_cache:
649 # Cache this setting object to the global cache
650 setting.save_to_cache()
652 return setting
654 @classmethod
655 def get_setting(cls, key, backup_value=None, **kwargs):
656 """Get the value of a particular setting.
658 If it does not exist, return the backup value (default = None)
659 """
660 if ( 660 ↛ 665line 660 didn't jump to line 665 because the condition on line 660 was never true
661 cls.CHECK_SETTING_KEY
662 and key not in cls.SETTINGS
663 and not key.startswith('_')
664 ):
665 logger.warning(
666 "get_setting: Setting key '%s' is not defined for class %s", key, cls
667 )
669 # If no backup value is specified, attempt to retrieve a "default" value
670 if backup_value is None:
671 backup_value = cls.get_setting_default(key, **kwargs)
673 setting = cls.get_setting_object(key, **kwargs)
675 if setting:
676 value = setting.value
678 # Cast to boolean if necessary
679 if setting.is_bool():
680 value = InvenTree.helpers.str2bool(value)
682 # Cast to integer if necessary
683 if setting.is_int():
684 try:
685 value = int(value)
686 except (ValueError, TypeError):
687 value = backup_value
689 else:
690 value = backup_value
692 return value
694 @classmethod
695 def set_setting(cls, key, value, change_user=None, create=True, **kwargs):
696 """Set the value of a particular setting. If it does not exist, option to create it.
698 Args:
699 key: settings key
700 value: New value
701 change_user: User object (must be staff member to update a core setting)
702 create: If True, create a new setting if the specified key does not exist.
703 """
704 if ( 704 ↛ 709line 704 didn't jump to line 709 because the condition on line 704 was never true
705 cls.CHECK_SETTING_KEY
706 and key not in cls.SETTINGS
707 and not key.startswith('_')
708 ):
709 logger.warning(
710 "set_setting: Setting key '%s' is not defined for class %s", key, cls
711 )
713 if change_user is not None and not change_user.is_staff: 713 ↛ 714line 713 didn't jump to line 714 because the condition on line 713 was never true
714 return
716 # Do not write to the database under certain conditions
717 if ( 717 ↛ 723line 717 didn't jump to line 723 because the condition on line 717 was never true
718 InvenTree.ready.isImportingData()
719 or InvenTree.ready.isRunningMigrations()
720 or InvenTree.ready.isRebuildingData()
721 or InvenTree.ready.isRunningBackup()
722 ): # pragma: no cover
723 return
725 attempts = int(kwargs.pop('attempts', 3))
727 filters = {
728 'key__iexact': key,
729 # Optionally filter by other keys
730 **cls.get_filters(**kwargs),
731 }
733 try:
734 setting = cls.objects.filter(**filters).first()
736 if not setting:
737 if create: 737 ↛ 740line 737 didn't jump to line 740 because the condition on line 737 was always true
738 setting = cls(key=key, **kwargs)
739 else:
740 return
742 except (OperationalError, ProgrammingError):
743 if not key.startswith('_'): 743 ↛ 744line 743 didn't jump to line 744 because the condition on line 743 was never true
744 logger.warning("Database is locked, cannot set setting '%s'", key)
745 # Likely the DB is locked - not much we can do here
746 return
747 except Exception as exc: # pragma: no cover
748 logger.exception(
749 "Error setting setting '%s' for %s: %s", key, cls, type(exc)
750 )
751 return
753 # Enforce standard boolean representation
754 if setting.is_bool(): 754 ↛ 755line 754 didn't jump to line 755 because the condition on line 754 was never true
755 value = InvenTree.helpers.str2bool(value)
757 try:
758 setting.value = str(value)
759 setting.save()
760 except ValidationError as exc:
761 # We need to know about validation errors
762 raise exc
763 except IntegrityError:
764 # Likely a race condition has caused a duplicate entry to be created
765 if attempts > 0:
766 # Try again
767 logger.info(
768 "Duplicate setting key '%s' for %s - trying again", key, cls
769 )
770 cls.set_setting(
771 key,
772 value,
773 change_user,
774 create=create,
775 attempts=attempts - 1,
776 **kwargs,
777 )
779 except (OperationalError, ProgrammingError):
780 logger.warning("Database is locked, cannot set setting '%s'", key)
781 # Likely the DB is locked - not much we can do here
782 except Exception as exc: # pragma: no cover
783 # Some other error
784 logger.exception(
785 "Error setting setting '%s' for %s: %s", key, cls, type(exc)
786 )
788 key = models.CharField(
789 max_length=50, blank=False, unique=False, help_text=_('Settings key')
790 )
792 value = models.CharField(
793 max_length=2000, blank=True, unique=False, help_text=_('Settings value')
794 )
796 @property
797 def name(self):
798 """Return name for setting."""
799 return self.__class__.get_setting_name(
800 self.key, **self.get_filters_for_instance()
801 )
803 @property
804 def default_value(self):
805 """Return default_value for setting."""
806 return self.__class__.get_setting_default(
807 self.key, **self.get_filters_for_instance()
808 )
810 @property
811 def description(self):
812 """Return description for setting."""
813 return self.__class__.get_setting_description(
814 self.key, **self.get_filters_for_instance()
815 )
817 @property
818 def units(self):
819 """Return units for setting."""
820 return self.__class__.get_setting_units(
821 self.key, **self.get_filters_for_instance()
822 )
824 def clean(self):
825 """If a validator (or multiple validators) are defined for a particular setting key, run them against the 'value' field."""
826 super().clean()
828 # Encode as native values
829 if self.is_int():
830 self.value = self.as_int(raise_error=True)
832 elif self.is_bool():
833 self.value = self.as_bool()
835 elif self.is_float(): 835 ↛ 836line 835 didn't jump to line 836 because the condition on line 835 was never true
836 self.value = self.as_float(raise_error=True)
838 validator = self.__class__.get_setting_validator(
839 self.key, **self.get_filters_for_instance()
840 )
842 if validator is not None:
843 self.run_validator(validator)
845 options = self.valid_options()
847 if options and self.value not in options: 847 ↛ 848line 847 didn't jump to line 848 because the condition on line 847 was never true
848 raise ValidationError(_('Chosen value is not a valid option'))
850 def run_validator(self, validator):
851 """Run a validator against the 'value' field for this InvenTreeSetting object."""
852 if validator is None: 852 ↛ 853line 852 didn't jump to line 853 because the condition on line 852 was never true
853 return
855 value = self.value
857 # Boolean validator
858 if validator is bool:
859 # Value must "look like" a boolean value
860 if InvenTree.helpers.is_bool(value): 860 ↛ 864line 860 didn't jump to line 864 because the condition on line 860 was always true
861 # Coerce into either "True" or "False"
862 value = InvenTree.helpers.str2bool(value)
863 else:
864 raise ValidationError({'value': _('Value must be a boolean value')})
866 # Integer validator
867 if validator is int:
868 try:
869 # Coerce into an integer value
870 value = int(value)
871 except (ValueError, TypeError):
872 raise ValidationError({'value': _('Value must be an integer value')})
874 # Floating point validator
875 if validator is float: 875 ↛ 876line 875 didn't jump to line 876 because the condition on line 875 was never true
876 try:
877 # Coerce into a floating point value
878 value = float(value)
879 except (ValueError, TypeError):
880 raise ValidationError({'value': _('Value must be a valid number')})
882 # If a list of validators is supplied, iterate through each one
883 if type(validator) in [list, tuple]:
884 for v in validator:
885 self.run_validator(v)
887 if callable(validator):
888 # We can accept function validators with a single argument
890 if self.is_bool():
891 value = self.as_bool()
893 elif self.is_int(): 893 ↛ 896line 893 didn't jump to line 896 because the condition on line 893 was always true
894 value = self.as_int()
896 elif self.is_float():
897 value = self.as_float()
899 try:
900 validator(value)
901 except ValidationError as e:
902 raise e
903 except Exception:
904 raise ValidationError({
905 'value': _('Value does not pass validation checks')
906 })
908 def validate_unique(self, exclude=None):
909 """Ensure that the key:value pair is unique. In addition to the base validators, this ensures that the 'key' is unique, using a case-insensitive comparison.
911 Note that sub-classes (UserSetting, PluginSetting) use other filters
912 to determine if the setting is 'unique' or not
913 """
914 super().validate_unique(exclude)
916 filters = {
917 'key__iexact': self.key,
918 # Optionally filter by other keys
919 **self.get_filters_for_instance(),
920 }
922 try:
923 # Check if a duplicate setting already exists
924 setting = self.__class__.objects.filter(**filters).exclude(id=self.id)
926 if setting.exists(): 926 ↛ 927line 926 didn't jump to line 927 because the condition on line 926 was never true
927 raise ValidationError({'key': _('Key string must be unique')})
929 except self.DoesNotExist:
930 pass
932 def choices(self):
933 """Return the available choices for this setting (or None if no choices are defined)."""
934 return self.__class__.get_setting_choices(
935 self.key, **self.get_filters_for_instance()
936 )
938 def valid_options(self):
939 """Return a list of valid options for this setting."""
940 choices = self.choices()
942 if not choices:
943 return None
945 return [opt[0] for opt in choices]
947 def is_choice(self):
948 """Check if this setting is a "choice" field."""
949 return (
950 self.__class__.get_setting_choices(
951 self.key, **self.get_filters_for_instance()
952 )
953 is not None
954 )
956 def as_choice(self):
957 """Render this setting as the "display" value of a choice field.
959 E.g. if the choices are:
960 [('A4', 'A4 paper'), ('A3', 'A3 paper')],
961 and the value is 'A4',
962 then display 'A4 paper'
963 """
964 choices = self.get_setting_choices(self.key, **self.get_filters_for_instance())
966 if not choices:
967 return self.value
969 for value, display in choices:
970 if value == self.value:
971 return display
973 return self.value
975 def is_model(self):
976 """Check if this setting references a model instance in the database."""
977 return self.model_name() is not None
979 def model_name(self) -> str:
980 """Return the model name associated with this setting."""
981 setting = self.get_setting_definition(
982 self.key, **self.get_filters_for_instance()
983 )
985 return setting.get('model', None)
987 def confirm(self) -> bool:
988 """Return if this setting requires confirmation on change."""
989 setting = self.get_setting_definition(
990 self.key, **self.get_filters_for_instance()
991 )
993 return setting.get('confirm', False)
995 def confirm_text(self) -> str:
996 """Return the confirmation text for this setting, if provided."""
997 setting = self.get_setting_definition(
998 self.key, **self.get_filters_for_instance()
999 )
1001 return setting.get('confirm_text', '')
1003 def model_filters(self) -> Optional[dict]:
1004 """Return the model filters associated with this setting."""
1005 setting = self.get_setting_definition(
1006 self.key, **self.get_filters_for_instance()
1007 )
1009 filters = setting.get('model_filters', None)
1011 if filters is not None and type(filters) is not dict: 1011 ↛ 1012line 1011 didn't jump to line 1012 because the condition on line 1011 was never true
1012 filters = None
1014 return filters
1016 def model_class(self):
1017 """Return the model class associated with this setting.
1019 If (and only if):
1020 - It has a defined 'model' parameter
1021 - The 'model' parameter is of the form app.model
1022 - The 'model' parameter has matches a known app model
1023 """
1024 model_name = self.model_name()
1026 if not model_name: 1026 ↛ 1030line 1026 didn't jump to line 1030 because the condition on line 1026 was always true
1027 return None
1029 # Enforce lower-case model name
1030 model_name = str(model_name).strip().lower()
1032 try:
1033 (app, mdl) = model_name.strip().split('.')
1034 except ValueError:
1035 logger.exception(
1036 "Invalid 'model' parameter for setting '%s': '%s'", self.key, model_name
1037 )
1038 return None
1040 app_models = apps.all_models.get(app, None)
1042 if app_models is None:
1043 logger.error(
1044 "Error retrieving model class '%s' for setting '%s' - no app named '%s'",
1045 model_name,
1046 self.key,
1047 app,
1048 )
1049 return None
1051 model = app_models.get(mdl, None)
1053 if model is None:
1054 logger.error(
1055 "Error retrieving model class '%s' for setting '%s' - no model named '%s'",
1056 model_name,
1057 self.key,
1058 mdl,
1059 )
1060 return None
1062 # Looks like we have found a model!
1063 return model
1065 def api_url(self):
1066 """Return the API url associated with the linked model, if provided, and valid!"""
1067 model_class = self.model_class()
1069 if model_class: 1069 ↛ 1071line 1069 didn't jump to line 1071 because the condition on line 1069 was never true
1070 # If a valid class has been found, see if it has registered an API URL
1071 try:
1072 return model_class.get_api_url()
1073 except Exception:
1074 pass
1076 # Some other model types are hard-coded
1077 hardcoded_models = {
1078 'auth.user': 'api-user-list',
1079 'auth.group': 'api-group-list',
1080 }
1082 model_table = (
1083 f'{model_class._meta.app_label}.{model_class._meta.model_name}'
1084 )
1086 if url := hardcoded_models[model_table]:
1087 return reverse(url)
1089 return None
1091 def is_bool(self):
1092 """Check if this setting is required to be a boolean value."""
1093 validator = self.__class__.get_setting_validator(
1094 self.key, **self.get_filters_for_instance()
1095 )
1097 return self.__class__.validator_is_bool(validator)
1099 def as_bool(self) -> bool:
1100 """Return the value of this setting converted to a boolean value.
1102 Warning: Only use on values where is_bool evaluates to true!
1103 """
1104 return InvenTree.helpers.str2bool(self.value)
1106 def setting_type(self):
1107 """Return the field type identifier for this setting object."""
1108 if self.is_bool():
1109 return 'boolean'
1111 elif self.is_int():
1112 return 'integer'
1114 elif self.is_model(): 1114 ↛ 1115line 1114 didn't jump to line 1115 because the condition on line 1114 was never true
1115 return 'related field'
1116 return 'string'
1118 @classmethod
1119 def validator_is_bool(cls, validator):
1120 """Return if validator is for bool."""
1121 if validator == bool:
1122 return True
1124 if type(validator) in [list, tuple]:
1125 for v in validator:
1126 if v == bool: 1126 ↛ 1127line 1126 didn't jump to line 1127 because the condition on line 1126 was never true
1127 return True
1129 return False
1131 def is_float(self):
1132 """Check if the setting is required to be a float value."""
1133 validator = self.__class__.get_setting_validator(
1134 self.key, **self.get_filters_for_instance()
1135 )
1137 return self.__class__.validator_is_float(validator)
1139 @classmethod
1140 def validator_is_float(cls, validator):
1141 """Return if validator is for float."""
1142 if validator == float: 1142 ↛ 1143line 1142 didn't jump to line 1143 because the condition on line 1142 was never true
1143 return True
1145 if type(validator) in [list, tuple]: 1145 ↛ 1146line 1145 didn't jump to line 1146 because the condition on line 1145 was never true
1146 for v in validator:
1147 if v == float:
1148 return True
1150 return False
1152 def as_float(self, raise_error: bool = False) -> float:
1153 """Return the value of this setting converted to a float value.
1155 If an error occurs, return the default value
1156 """
1157 try:
1158 value = float(self.value)
1159 except (ValueError, TypeError):
1160 if raise_error:
1161 raise ValidationError('Provided value is not a valid float')
1162 value = self.default_value
1164 return value
1166 def is_int(self):
1167 """Check if the setting is required to be an integer value."""
1168 validator = self.__class__.get_setting_validator(
1169 self.key, **self.get_filters_for_instance()
1170 )
1172 return self.__class__.validator_is_int(validator)
1174 @classmethod
1175 def validator_is_int(cls, validator):
1176 """Return if validator is for int."""
1177 if validator == int:
1178 return True
1180 if type(validator) in [list, tuple]:
1181 for v in validator: 1181 ↛ 1185line 1181 didn't jump to line 1185 because the loop on line 1181 didn't complete
1182 if v == int: 1182 ↛ 1181line 1182 didn't jump to line 1181 because the condition on line 1182 was always true
1183 return True
1185 return False
1187 def as_int(self, raise_error: bool = False) -> int:
1188 """Return the value of this setting converted to a boolean value.
1190 If an error occurs, return the default value
1191 """
1192 try:
1193 value = int(self.value)
1194 except (ValueError, TypeError):
1195 if raise_error:
1196 raise ValidationError('Provided value is not a valid integer')
1197 value = self.default_value
1199 return value
1201 @classmethod
1202 def is_protected(cls, key, **kwargs):
1203 """Check if the setting value is protected."""
1204 setting = cls.get_setting_definition(key, **cls.get_filters(**kwargs))
1206 return setting.get('protected', False)
1208 @property
1209 def protected(self):
1210 """Returns if setting is protected from rendering."""
1211 return self.__class__.is_protected(self.key, **self.get_filters_for_instance())
1213 @classmethod
1214 def is_required(cls, key, **kwargs):
1215 """Check if this setting value is required."""
1216 setting = cls.get_setting_definition(key, **cls.get_filters(**kwargs))
1218 return setting.get('required', False)
1220 @property
1221 def required(self):
1222 """Returns if setting is required."""
1223 return self.__class__.is_required(self.key, **self.get_filters_for_instance())
1226class InvenTreeSetting(BaseInvenTreeSetting):
1227 """An InvenTreeSetting object is a key:value pair used for storing single values (e.g. one-off settings values).
1229 The class provides a way of retrieving the value for a particular key,
1230 even if that key does not exist.
1231 """
1233 from common.setting.system import SYSTEM_SETTINGS
1235 SETTINGS: dict[str, InvenTreeSettingsKeyType] = SYSTEM_SETTINGS
1237 CHECK_SETTING_KEY = True
1239 class Meta:
1240 """Meta options for InvenTreeSetting."""
1242 verbose_name = 'InvenTree Setting'
1243 verbose_name_plural = 'InvenTree Settings'
1245 def save(self, *args, **kwargs):
1246 """When saving a global setting, check to see if it requires a server restart.
1248 If so, set the "SERVER_RESTART_REQUIRED" setting to True
1249 """
1250 overrides = global_setting_overrides()
1252 # If an override is specified for this setting, use that value
1253 if self.key in overrides: 1253 ↛ 1254line 1253 didn't jump to line 1254 because the condition on line 1253 was never true
1254 self.value = overrides[self.key]
1256 super().save()
1258 if self.requires_restart() and not InvenTree.ready.isImportingData(): 1258 ↛ 1259line 1258 didn't jump to line 1259 because the condition on line 1258 was never true
1259 InvenTreeSetting.set_setting('SERVER_RESTART_REQUIRED', True, None)
1261 @classmethod
1262 def get_setting_default(cls, key, **kwargs):
1263 """Return the default value a particular setting."""
1264 overrides = global_setting_overrides()
1266 if key in overrides:
1267 # If an override is specified for this setting, use that value
1268 return overrides[key]
1270 return super().get_setting_default(key, **kwargs)
1272 @classmethod
1273 def get_setting(cls, key, backup_value=None, **kwargs):
1274 """Get the value of a particular setting.
1276 If it does not exist, return the backup value (default = None)
1277 """
1278 overrides = global_setting_overrides()
1280 if key in overrides: 1280 ↛ 1282line 1280 didn't jump to line 1282 because the condition on line 1280 was never true
1281 # If an override is specified for this setting, use that value
1282 return overrides[key]
1284 return super().get_setting(key, backup_value=backup_value, **kwargs)
1286 """
1287 Dict of all global settings values:
1289 The key of each item is the name of the value as it appears in the database.
1291 Each global setting has the following parameters:
1293 - name: Translatable string name of the setting (required)
1294 - description: Translatable string description of the setting (required)
1295 - default: Default value (optional)
1296 - units: Units of the particular setting (optional)
1297 - validator: Validation function for the setting (optional)
1299 The keys must be upper-case
1300 """
1302 typ = 'inventree'
1304 key = models.CharField(
1305 max_length=50, blank=False, unique=True, help_text=_('Settings key')
1306 )
1308 def to_native_value(self):
1309 """Return the "pythonic" value, e.g. convert "True" to True, and "1" to 1."""
1310 return self.__class__.get_setting(self.key)
1312 def requires_restart(self):
1313 """Return True if this setting requires a server restart after changing."""
1314 options = InvenTreeSetting.SETTINGS.get(self.key, None)
1316 if options:
1317 return options.get('requires_restart', False)
1318 return False
1321class InvenTreeUserSetting(BaseInvenTreeSetting):
1322 """An InvenTreeSetting object with a user context."""
1324 import common.setting.user
1326 SETTINGS = common.setting.user.USER_SETTINGS
1328 CHECK_SETTING_KEY = True
1330 class Meta:
1331 """Meta options for InvenTreeUserSetting."""
1333 verbose_name = 'InvenTree User Setting'
1334 verbose_name_plural = 'InvenTree User Settings'
1335 constraints = [
1336 models.UniqueConstraint(fields=['key', 'user'], name='unique key and user')
1337 ]
1339 typ = 'user'
1340 extra_unique_fields = ['user']
1342 key = models.CharField(
1343 max_length=50, blank=False, unique=False, help_text=_('Settings key')
1344 )
1346 user = models.ForeignKey(
1347 User,
1348 on_delete=models.CASCADE,
1349 blank=True,
1350 null=True,
1351 verbose_name=_('User'),
1352 help_text=_('User'),
1353 )
1355 def to_native_value(self):
1356 """Return the "pythonic" value, e.g. convert "True" to True, and "1" to 1."""
1357 return self.__class__.get_setting(self.key, user=self.user)
1360class PriceBreak(MetaMixin):
1361 """Represents a PriceBreak model."""
1363 class Meta:
1364 """Define this as abstract -> no DB entry is created."""
1366 abstract = True
1368 quantity = InvenTree.fields.RoundingDecimalField(
1369 max_digits=15,
1370 decimal_places=5,
1371 default=1,
1372 validators=[MinValueValidator(1)],
1373 verbose_name=_('Quantity'),
1374 help_text=_('Price break quantity'),
1375 )
1377 price = InvenTree.fields.InvenTreeModelMoneyField(
1378 max_digits=19,
1379 decimal_places=6,
1380 null=True,
1381 verbose_name=_('Price'),
1382 help_text=_('Unit price at specified quantity'),
1383 )
1385 def convert_to(self, currency_code):
1386 """Convert the unit-price at this price break to the specified currency code.
1388 Args:
1389 currency_code: The currency code to convert to (e.g "USD" or "AUD")
1390 """
1391 try:
1392 converted = convert_money(self.price, currency_code)
1393 except MissingRate:
1394 logger.warning(
1395 'No currency conversion rate available for %s -> %s',
1396 self.price_currency,
1397 currency_code,
1398 )
1399 return self.price.amount
1401 return converted.amount
1404class VerificationMethod(Enum):
1405 """Class to hold method references."""
1407 NONE = 0
1408 TOKEN = 1
1409 HMAC = 2
1412class WebhookEndpoint(models.Model):
1413 """Defines a Webhook endpoint.
1415 Attributes:
1416 endpoint_id: Path to the webhook,
1417 name: Name of the webhook,
1418 active: Is this webhook active?,
1419 user: User associated with webhook,
1420 token: Token for sending a webhook,
1421 secret: Shared secret for HMAC verification,
1422 """
1424 # Token
1425 TOKEN_NAME = 'Token'
1426 VERIFICATION_METHOD = VerificationMethod.NONE
1428 MESSAGE_OK = 'Message was received.'
1429 MESSAGE_TOKEN_ERROR = 'Incorrect token in header.'
1431 endpoint_id = models.CharField(
1432 max_length=255,
1433 verbose_name=_('Endpoint'),
1434 help_text=_('Endpoint at which this webhook is received'),
1435 default=uuid.uuid4,
1436 editable=False,
1437 )
1439 name = models.CharField(
1440 max_length=255,
1441 blank=True,
1442 null=True,
1443 verbose_name=_('Name'),
1444 help_text=_('Name for this webhook'),
1445 )
1447 active = models.BooleanField(
1448 default=True, verbose_name=_('Active'), help_text=_('Is this webhook active')
1449 )
1451 user = models.ForeignKey(
1452 User,
1453 on_delete=models.SET_NULL,
1454 blank=True,
1455 null=True,
1456 verbose_name=_('User'),
1457 help_text=_('User'),
1458 )
1460 token = models.CharField(
1461 max_length=255,
1462 blank=True,
1463 null=True,
1464 verbose_name=_('Token'),
1465 help_text=_('Token for access'),
1466 default=uuid.uuid4,
1467 )
1469 secret = models.CharField(
1470 max_length=255,
1471 blank=True,
1472 null=True,
1473 verbose_name=_('Secret'),
1474 help_text=_('Shared secret for HMAC'),
1475 )
1477 # To be overridden
1479 def init(self, request, *args, **kwargs):
1480 """Set verification method.
1482 Args:
1483 request: Original request object.
1484 """
1485 self.verify = self.VERIFICATION_METHOD
1487 def process_webhook(self):
1488 """Process the webhook incoming.
1490 This does not deal with the data itself - that happens in process_payload.
1491 Do not touch or pickle data here - it was not verified to be safe.
1492 """
1493 if self.token:
1494 self.verify = VerificationMethod.TOKEN
1495 if self.secret:
1496 self.verify = VerificationMethod.HMAC
1497 return True
1499 def validate_token(self, payload, headers, request):
1500 """Make sure that the provided token (if any) confirms to the setting for this endpoint.
1502 This can be overridden to create your own token validation method.
1503 """
1504 token = headers.get(self.TOKEN_NAME, '')
1506 # no token
1507 if self.verify == VerificationMethod.NONE:
1508 # do nothing as no method was chosen
1509 pass
1511 # static token
1512 elif self.verify == VerificationMethod.TOKEN:
1513 if not compare_digest(token, self.token):
1514 raise PermissionDenied(self.MESSAGE_TOKEN_ERROR)
1516 # hmac token
1517 elif self.verify == VerificationMethod.HMAC:
1518 digest = hmac.new(
1519 self.secret.encode('utf-8'), request.body, hashlib.sha256
1520 ).digest()
1521 computed_hmac = base64.b64encode(digest)
1522 if not hmac.compare_digest(computed_hmac, token.encode('utf-8')):
1523 raise PermissionDenied(self.MESSAGE_TOKEN_ERROR)
1525 return True
1527 def save_data(self, payload=None, headers=None, request=None):
1528 """Safes payload to database.
1530 Args:
1531 payload (optional): Payload that was send along. Defaults to None.
1532 headers (optional): Headers that were send along. Defaults to None.
1533 request (optional): Original request object. Defaults to None.
1534 """
1535 return WebhookMessage.objects.create(
1536 host=request.get_host() if request else '',
1537 header=json.dumps(dict(headers.items())) if headers else None,
1538 body=payload,
1539 endpoint=self,
1540 )
1542 def process_payload(self, message, payload=None, headers=None) -> bool:
1543 """Process a payload.
1545 Args:
1546 message: DB entry for this message mm
1547 payload (optional): Payload that was send along. Defaults to None.
1548 headers (optional): Headers that were included. Defaults to None.
1550 Returns:
1551 bool: Was the message processed
1552 """
1553 return True
1555 def get_return(self, payload=None, headers=None, request=None) -> str:
1556 """Returns the message that should be returned to the endpoint caller.
1558 Args:
1559 payload (optional): Payload that was send along. Defaults to None.
1560 headers (optional): Headers that were send along. Defaults to None.
1561 request (optional): Original request object. Defaults to None.
1563 Returns:
1564 str: Message for caller.
1565 """
1566 return self.MESSAGE_OK
1569class WebhookMessage(models.Model):
1570 """Defines a webhook message.
1572 Attributes:
1573 message_id: Unique identifier for this message,
1574 host: Host from which this message was received,
1575 header: Header of this message,
1576 body: Body of this message,
1577 endpoint: Endpoint on which this message was received,
1578 worked_on: Was the work on this message finished?
1579 """
1581 message_id = models.UUIDField(
1582 verbose_name=_('Message ID'),
1583 help_text=_('Unique identifier for this message'),
1584 primary_key=True,
1585 default=uuid.uuid4,
1586 editable=False,
1587 )
1589 host = models.CharField(
1590 max_length=255,
1591 verbose_name=_('Host'),
1592 help_text=_('Host from which this message was received'),
1593 editable=False,
1594 )
1596 header = models.CharField(
1597 max_length=255,
1598 blank=True,
1599 null=True,
1600 verbose_name=_('Header'),
1601 help_text=_('Header of this message'),
1602 editable=False,
1603 )
1605 body = models.JSONField(
1606 blank=True,
1607 null=True,
1608 verbose_name=_('Body'),
1609 help_text=_('Body of this message'),
1610 editable=False,
1611 )
1613 endpoint = models.ForeignKey(
1614 WebhookEndpoint,
1615 on_delete=models.SET_NULL,
1616 blank=True,
1617 null=True,
1618 verbose_name=_('Endpoint'),
1619 help_text=_('Endpoint on which this message was received'),
1620 )
1622 worked_on = models.BooleanField(
1623 default=False,
1624 verbose_name=_('Worked on'),
1625 help_text=_('Was the work on this message finished?'),
1626 )
1629class NotificationEntry(MetaMixin):
1630 """A NotificationEntry records the last time a particular notification was sent out.
1632 It is recorded to ensure that notifications are not sent out "too often" to users.
1634 Attributes:
1635 - key: A text entry describing the notification e.g. 'part.notify_low_stock'
1636 - uid: An (optional) numerical ID for a particular instance
1637 - date: The last time this notification was sent
1638 """
1640 class Meta:
1641 """Meta options for NotificationEntry."""
1643 unique_together = [('key', 'uid')]
1645 key = models.CharField(max_length=250, blank=False)
1647 uid = models.IntegerField()
1649 @classmethod
1650 def check_recent(cls, key: str, uid: int, delta: timedelta):
1651 """Test if a particular notification has been sent in the specified time period."""
1652 since = InvenTree.helpers.current_date() - delta
1654 entries = cls.objects.filter(key=key, uid=uid, updated__gte=since)
1656 return entries.exists()
1658 @classmethod
1659 def notify(cls, key: str, uid: int):
1660 """Notify the database that a particular notification has been sent out."""
1661 entry, _ = cls.objects.get_or_create(key=key, uid=uid)
1663 entry.save()
1666class NotificationMessage(models.Model):
1667 """A NotificationMessage is a message sent to a particular user, notifying them of some important information.
1669 Notification messages can be generated by a variety of sources.
1671 Attributes:
1672 target_object: The 'target' of the notification message
1673 source_object: The 'source' of the notification message
1674 """
1676 # generic link to target
1677 target_content_type = models.ForeignKey(
1678 ContentType, on_delete=models.CASCADE, related_name='notification_target'
1679 )
1681 target_object_id = models.CharField(max_length=255)
1683 target_object = GenericForeignKey('target_content_type', 'target_object_id')
1685 # generic link to source
1686 source_content_type = models.ForeignKey(
1687 ContentType,
1688 on_delete=models.SET_NULL,
1689 related_name='notification_source',
1690 null=True,
1691 blank=True,
1692 )
1694 source_object_id = models.CharField(max_length=255, null=True, blank=True)
1696 source_object = GenericForeignKey('source_content_type', 'source_object_id')
1698 # user that receives the notification
1699 user = models.ForeignKey(
1700 User,
1701 on_delete=models.CASCADE,
1702 verbose_name=_('User'),
1703 help_text=_('User'),
1704 null=True,
1705 blank=True,
1706 )
1708 category = models.CharField(max_length=250, blank=False)
1710 name = models.CharField(max_length=250, blank=False)
1712 message = models.CharField(max_length=250, blank=True, null=True)
1714 creation = models.DateTimeField(auto_now_add=True)
1716 read = models.BooleanField(default=False)
1718 @staticmethod
1719 def get_api_url():
1720 """Return API endpoint."""
1721 return reverse('api-notifications-list')
1723 def age(self) -> int:
1724 """Age of the message in seconds."""
1725 # Add timezone information if TZ is enabled (in production mode mostly)
1726 delta = now() - (
1727 self.creation.replace(tzinfo=timezone.utc)
1728 if django_settings.USE_TZ
1729 else self.creation
1730 )
1731 return delta.seconds
1733 def age_human(self) -> str:
1734 """Humanized age."""
1735 return naturaltime(self.creation)
1738class NewsFeedEntry(models.Model):
1739 """A NewsFeedEntry represents an entry on the RSS/Atom feed that is generated for InvenTree news.
1741 Attributes:
1742 - feed_id: Unique id for the news item
1743 - title: Title for the news item
1744 - link: Link to the news item
1745 - published: Date of publishing of the news item
1746 - author: Author of news item
1747 - summary: Summary of the news items content
1748 - read: Was this item already by a superuser?
1749 """
1751 feed_id = models.CharField(verbose_name=_('Id'), unique=True, max_length=250)
1753 title = models.CharField(verbose_name=_('Title'), max_length=250)
1755 link = models.URLField(verbose_name=_('Link'), max_length=250)
1757 published = models.DateTimeField(verbose_name=_('Published'), max_length=250)
1759 author = models.CharField(verbose_name=_('Author'), max_length=250)
1761 summary = models.CharField(verbose_name=_('Summary'), max_length=250)
1763 read = models.BooleanField(
1764 verbose_name=_('Read'), help_text=_('Was this news item read?'), default=False
1765 )
1768def rename_notes_image(instance, filename):
1769 """Function for renaming uploading image file. Will store in the 'notes' directory."""
1770 fname = os.path.basename(filename)
1771 return os.path.join('notes', fname)
1774class NotesImage(models.Model):
1775 """Model for storing uploading images for the 'notes' fields of various models.
1777 Simply stores the image file, for use in the 'notes' field (of any models which support markdown).
1778 """
1780 image = models.ImageField(
1781 upload_to=rename_notes_image, verbose_name=_('Image'), help_text=_('Image file')
1782 )
1784 user = models.ForeignKey(User, on_delete=models.SET_NULL, null=True, blank=True)
1786 date = models.DateTimeField(auto_now_add=True)
1788 model_type = models.CharField(
1789 max_length=100,
1790 blank=True,
1791 null=True,
1792 validators=[common.validators.validate_notes_model_type],
1793 help_text=_('Target model type for this image'),
1794 )
1796 model_id = models.IntegerField(
1797 help_text=_('Target model ID for this image'),
1798 blank=True,
1799 null=True,
1800 default=None,
1801 )
1804class CustomUnit(models.Model):
1805 """Model for storing custom physical unit definitions.
1807 Model Attributes:
1808 name: Name of the unit
1809 definition: Definition of the unit
1810 symbol: Symbol for the unit (e.g. 'm' for 'metre') (optional)
1812 Refer to the pint documentation for further information on unit definitions.
1813 https://pint.readthedocs.io/en/stable/advanced/defining.html
1814 """
1816 class Meta:
1817 """Class meta options."""
1819 verbose_name = _('Custom Unit')
1821 def fmt_string(self):
1822 """Construct a unit definition string e.g. 'dog_year = 52 * day = dy'."""
1823 fmt = f'{self.name} = {self.definition}'
1825 if self.symbol:
1826 fmt += f' = {self.symbol}'
1828 return fmt
1830 def validate_unique(self, exclude=None) -> None:
1831 """Ensure that the custom unit is unique."""
1832 super().validate_unique(exclude)
1834 if self.symbol and ( 1834 ↛ 1837line 1834 didn't jump to line 1837 because the condition on line 1834 was never true
1835 CustomUnit.objects.filter(symbol=self.symbol).exclude(pk=self.pk).exists()
1836 ):
1837 raise ValidationError({'symbol': _('Unit symbol must be unique')})
1839 def clean(self):
1840 """Validate that the provided custom unit is indeed valid."""
1841 super().clean()
1843 from InvenTree.conversion import get_unit_registry
1845 registry = get_unit_registry()
1847 # Check that the 'name' field is valid
1848 self.name = self.name.strip()
1850 # Cannot be zero length
1851 if not self.name.isidentifier():
1852 raise ValidationError({'name': _('Unit name must be a valid identifier')})
1854 self.definition = self.definition.strip()
1856 # Check that the 'definition' is valid, by itself
1857 try:
1858 registry.Quantity(self.definition)
1859 except Exception as exc:
1860 raise ValidationError({'definition': str(exc)})
1862 # Finally, test that the entire custom unit definition is valid
1863 try:
1864 registry.define(self.fmt_string())
1865 except Exception as exc:
1866 raise ValidationError(str(exc))
1868 name = models.CharField(
1869 max_length=50,
1870 verbose_name=_('Name'),
1871 help_text=_('Unit name'),
1872 unique=True,
1873 blank=False,
1874 )
1876 symbol = models.CharField(
1877 max_length=10,
1878 verbose_name=_('Symbol'),
1879 help_text=_('Optional unit symbol'),
1880 blank=True,
1881 )
1883 definition = models.CharField(
1884 max_length=50,
1885 verbose_name=_('Definition'),
1886 help_text=_('Unit definition'),
1887 blank=False,
1888 )
1891@receiver(post_save, sender=CustomUnit, dispatch_uid='custom_unit_saved')
1892@receiver(post_delete, sender=CustomUnit, dispatch_uid='custom_unit_deleted')
1893def after_custom_unit_updated(sender, instance, **kwargs):
1894 """Callback when a custom unit is updated or deleted."""
1895 # Force reload of the unit registry
1896 from InvenTree.conversion import reload_unit_registry
1898 reload_unit_registry()
1901def rename_attachment(instance, filename: str):
1902 """Callback function to rename an uploaded attachment file.
1904 Args:
1905 instance (Attachment): The Attachment instance for which the file is being renamed.
1906 filename (str): The original filename of the uploaded file.
1908 Returns:
1909 str: The new filename for the uploaded file, e.g. 'attachments/<model_type>/<model_id>/<filename>'.
1910 """
1911 # Remove any illegal characters from the filename
1912 illegal_chars = '\'"\\`~#|!@#$%^&*()[]{}<>?;:+=,'
1914 for c in illegal_chars:
1915 filename = filename.replace(c, '')
1917 filename = os.path.basename(filename)
1919 # Generate a new filename for the attachment
1920 return os.path.join(
1921 'attachments', str(instance.model_type), str(instance.model_id), filename
1922 )
1925class Attachment(
1926 InvenTree.models.MetadataMixin,
1927 InvenTree.models.InvenTreeTagsMixin,
1928 InvenTree.models.InvenTreeModel,
1929):
1930 """Class which represents an uploaded file attachment.
1932 An attachment can be either an uploaded file, or an external URL.
1934 Attributes:
1935 model_type: The type of model to which this attachment is linked
1936 model_id: The ID of the model to which this attachment is linked
1937 attachment: The uploaded file
1938 url: An external URL
1939 thumbnail: A generated thumbnail for the uploaded file (if applicable)
1940 is_image: True if this attachment is a valid image file
1941 comment: A comment or description for the attachment
1942 user: The user who uploaded the attachment
1943 upload_date: The date the attachment was uploaded
1944 file_size: The size of the uploaded file
1945 metadata: Arbitrary metadata for the attachment (inherit from MetadataMixin)
1946 tags: Tags for the attachment
1947 """
1949 THUMBNAIL_SIZE = 256
1951 class Meta:
1952 """Metaclass options."""
1954 verbose_name = _('Attachment')
1956 class ModelChoices(RenderChoices):
1957 """Model choices for attachments."""
1959 choice_fnc = common.validators.attachment_model_options
1961 def delete(self, *args, **kwargs):
1962 """Custom delete method for the Attachment model.
1964 - Ensure that the attached file is deleted from storage when the database entry is removed
1965 """
1966 attachment = self.attachment
1967 thumbnail = self.thumbnail
1969 super().delete(*args, **kwargs)
1971 # Delete the associated files from storage (if they exist)W
1972 if attachment and default_storage.exists(attachment.name):
1973 try:
1974 # Remove the attached file from storage
1975 default_storage.delete(attachment.name)
1976 except Exception: # pragma: no cover
1977 pass
1979 if thumbnail and default_storage.exists(thumbnail.name):
1980 try:
1981 # Remove the thumbnail file from storage
1982 default_storage.delete(thumbnail.name)
1983 except Exception: # pragma: no cover
1984 pass
1986 def save(self, *args, **kwargs):
1987 """Custom 'save' method for the Attachment model.
1989 - Record the file size of the uploaded attachment (if applicable)
1990 - Ensure that the 'content_type' and 'object_id' fields are set
1991 - Run extra validations
1992 """
1993 import common.tasks
1995 rebuild = kwargs.pop('rebuild', True)
1997 # Either 'attachment' or 'link' must be specified!
1998 if not self.attachment and not self.link: 1998 ↛ 2004line 1998 didn't jump to line 2004 because the condition on line 1998 was always true
1999 raise ValidationError({
2000 'attachment': _('Missing file'),
2001 'link': _('Missing external link'),
2002 })
2004 if self.attachment:
2005 if self.attachment.name.lower().endswith('.svg'):
2006 self.attachment.file.file = self.clean_svg(self.attachment)
2007 else:
2008 self.file_size = 0
2010 super().save(*args, **kwargs)
2012 # Update file size
2013 if self.file_size == 0 and self.attachment:
2014 # Get file size
2015 if default_storage.exists(self.attachment.name):
2016 try:
2017 self.file_size = default_storage.size(self.attachment.name)
2018 except Exception:
2019 pass
2021 if self.file_size != 0:
2022 super().save()
2024 # Offload a background task to update the thumbnail for this attachment
2025 if rebuild:
2026 InvenTree.tasks.offload_task(
2027 common.tasks.rebuild_attachment, self.pk, group='attachments'
2028 )
2030 def clean_svg(self, field):
2031 """Sanitize SVG file before saving."""
2032 cleaned = sanitize_svg(field.file.read())
2033 return BytesIO(bytes(cleaned, 'utf8'))
2035 def __str__(self):
2036 """Human name for attachment."""
2037 if self.attachment is not None:
2038 return os.path.basename(self.attachment.name)
2039 return str(self.link)
2041 def validate_rename(self, filename: str):
2042 """Validate that the provided filename is valid, for renaming an attachment."""
2043 filename = filename.strip()
2045 if not self.attachment:
2046 raise ValidationError(_('No file attached to rename'))
2048 if not filename:
2049 raise ValidationError(_('Filename cannot be empty'))
2051 try:
2052 validate_file_name(filename, allow_relative_path=False)
2053 except SuspiciousFileOperation:
2054 raise ValidationError(_('Invalid filename'))
2056 current_ext = os.path.splitext(self.attachment.name)[1]
2057 new_ext = os.path.splitext(filename)[1]
2059 if current_ext.lower() != new_ext.lower():
2060 raise ValidationError(_('Cannot change file extension'))
2062 def rename(self, filename: str):
2063 """Rename the attached file."""
2064 self.validate_rename(filename)
2066 old_path = Path(self.attachment.name)
2067 new_path = old_path.parent / filename
2069 if old_path == new_path: # pragma: no cover
2070 # No change in filename
2071 return
2073 if not new_path.is_relative_to(old_path.parent): # pragma: no cover
2074 raise ValidationError(_('Invalid filename'))
2076 new_path = new_path.as_posix()
2078 if default_storage.exists(new_path):
2079 raise ValidationError(_('A file with this name already exists'))
2081 # Create a new file with the new name, and delete the old file
2082 new_path = default_storage.save(new_path, self.attachment.file)
2084 # Ensure that the new file exists
2085 if not default_storage.exists(new_path): # pragma: no cover
2086 raise ValidationError(_('Failed to save renamed file'))
2088 # Update the database file path
2089 self.attachment.name = new_path
2090 self.save()
2092 # Remove the old path
2093 default_storage.delete(old_path)
2095 model_type = models.CharField(
2096 max_length=100,
2097 validators=[common.validators.validate_attachment_model_type],
2098 verbose_name=_('Model type'),
2099 help_text=_('Target model type for image'),
2100 )
2102 model_id = models.PositiveIntegerField()
2104 attachment = models.FileField(
2105 upload_to=rename_attachment,
2106 verbose_name=_('Attachment'),
2107 validators=[common.validators.validate_attachment_file],
2108 help_text=_('Select file to attach'),
2109 blank=True,
2110 null=True,
2111 )
2113 thumbnail = models.ImageField(
2114 verbose_name=_('Thumbnail'),
2115 help_text=_('Thumbnail image for this attachment'),
2116 blank=True,
2117 null=True,
2118 )
2120 link = InvenTree.fields.InvenTreeURLField(
2121 blank=True,
2122 null=True,
2123 verbose_name=_('Link'),
2124 help_text=_('Link to external URL'),
2125 max_length=2000,
2126 )
2128 comment = models.CharField(
2129 blank=True,
2130 max_length=250,
2131 verbose_name=_('Comment'),
2132 help_text=_('Attachment comment'),
2133 )
2135 upload_user = models.ForeignKey(
2136 User,
2137 on_delete=models.SET_NULL,
2138 blank=True,
2139 null=True,
2140 verbose_name=_('User'),
2141 help_text=_('User'),
2142 )
2144 upload_date = models.DateField(
2145 auto_now_add=True,
2146 null=True,
2147 blank=True,
2148 verbose_name=_('Upload date'),
2149 help_text=_('Date the file was uploaded'),
2150 )
2152 is_image = models.BooleanField(
2153 default=False,
2154 verbose_name=_('Is image'),
2155 help_text=_('True if this attachment is a valid image file'),
2156 )
2158 file_size = models.PositiveIntegerField(
2159 default=0, verbose_name=_('File size'), help_text=_('File size in bytes')
2160 )
2162 @property
2163 def basename(self):
2164 """Base name/path for attachment."""
2165 if self.attachment:
2166 return os.path.basename(self.attachment.name)
2167 return None
2169 def fully_qualified_url(self):
2170 """Return a 'fully qualified' URL for this attachment.
2172 - If the attachment is a link to an external resource, return the link
2173 - If the attachment is an uploaded file, return the fully qualified media URL
2174 """
2175 if self.link:
2176 return self.link
2178 if self.attachment:
2179 import InvenTree.helpers_model
2181 media_url = InvenTree.helpers.getMediaUrl(self.attachment)
2182 return InvenTree.helpers_model.construct_absolute_url(media_url)
2184 return ''
2186 def check_permission(self, permission, user):
2187 """Check if the user has the required permission for this attachment."""
2188 from InvenTree.models import InvenTreeAttachmentMixin
2190 model_class = common.validators.attachment_model_class_from_label(
2191 self.model_type
2192 )
2194 if not issubclass(model_class, InvenTreeAttachmentMixin):
2195 raise ValidationError(_('Invalid model type specified for attachment'))
2197 return model_class.check_related_permission(permission, user)
2199 def check_is_image(self) -> bool:
2200 """Check if the attached file is an image.
2202 We consider it a valid image if:
2204 - The file exists in storage
2205 - The file can be opened and verified by the PIL library
2207 """
2208 if not self.attachment:
2209 return False
2211 if not self.attachment.name:
2212 return False
2214 try:
2215 if not default_storage.exists(self.attachment.name):
2216 return False
2217 except Exception:
2218 return False
2220 img_data = default_storage.open(self.attachment.name).read()
2222 try:
2223 Image.open(BytesIO(img_data)).verify()
2224 return True
2225 except Exception:
2226 return False
2228 def generate_thumbnail(self):
2229 """Generate a thumbnail for the attached image."""
2230 # Remove any existing thumbnail
2231 if self.thumbnail:
2232 self.thumbnail.delete(save=False)
2234 if not self.attachment:
2235 return
2237 if not self.attachment.name or not default_storage.exists(self.attachment.name):
2238 return
2240 # TODO: Offload to plugins, for creating custom thumbnails for different file types
2241 # TODO: If a plugin provides a thumbnail, return early
2243 # Default action is to generate a thumbnail for image files
2244 try:
2245 img_data = default_storage.open(self.attachment.name).read()
2246 except Exception:
2247 # No file found, or file cannot be read - cannot generate thumbnail
2248 return
2250 try:
2251 img = Image.open(BytesIO(img_data))
2252 img.thumbnail((self.THUMBNAIL_SIZE, self.THUMBNAIL_SIZE))
2253 thumb_io = BytesIO()
2254 img.save(thumb_io, format='PNG')
2255 thumb_io.seek(0)
2257 thumb_name = f'thumb_{os.path.basename(self.attachment.name)}'
2258 self.thumbnail.save(thumb_name, ContentFile(thumb_io.read()), save=False)
2259 except Exception:
2260 pass
2263class InvenTreeCustomUserStateModel(models.Model):
2264 """Custom model to extends any registered state with extra custom, user defined states.
2266 Fields:
2267 reference_status: Status set that is extended with this custom state
2268 logical_key: State logical key that is equal to this custom state in business logic
2269 key: Numerical value that will be saved in the models database
2270 name: Name of the state (must be uppercase and a valid variable identifier)
2271 label: Label that will be displayed in the frontend (human readable)
2272 color: Color that will be displayed in the frontend
2274 """
2276 class Meta:
2277 """Metaclass options for this mixin."""
2279 verbose_name = _('Custom State')
2280 verbose_name_plural = _('Custom States')
2281 unique_together = [('reference_status', 'key'), ('reference_status', 'name')]
2283 reference_status = models.CharField(
2284 max_length=250,
2285 verbose_name=_('Reference Status Set'),
2286 help_text=_('Status set that is extended with this custom state'),
2287 )
2289 logical_key = models.IntegerField(
2290 verbose_name=_('Logical Key'),
2291 help_text=_(
2292 'State logical key that is equal to this custom state in business logic'
2293 ),
2294 )
2296 key = models.IntegerField(
2297 verbose_name=_('Value'),
2298 help_text=_('Numerical value that will be saved in the models database'),
2299 )
2301 name = models.CharField(
2302 max_length=250,
2303 verbose_name=_('Name'),
2304 help_text=_('Name of the state'),
2305 validators=[
2306 common.validators.validate_uppercase,
2307 common.validators.validate_variable_string,
2308 ],
2309 )
2311 label = models.CharField(
2312 max_length=250,
2313 verbose_name=_('Label'),
2314 help_text=_('Label that will be displayed in the frontend'),
2315 )
2317 color = models.CharField(
2318 max_length=10,
2319 choices=state_color_mappings(),
2320 default=ColorEnum.secondary.value,
2321 verbose_name=_('Color'),
2322 help_text=_('Color that will be displayed in the frontend'),
2323 )
2325 model = models.ForeignKey(
2326 ContentType,
2327 on_delete=models.SET_NULL,
2328 null=True,
2329 blank=True,
2330 verbose_name=_('Model'),
2331 help_text=_('Model this state is associated with'),
2332 )
2334 def __str__(self) -> str:
2335 """Return string representation of the custom state."""
2336 return f'{self.model.name} ({self.reference_status}): {self.name} | {self.key} ({self.logical_key})'
2338 def save(self, *args, **kwargs) -> None:
2339 """Ensure that the custom state is valid before saving."""
2340 self.clean()
2341 return super().save(*args, **kwargs)
2343 def clean(self) -> None:
2344 """Validate custom state data."""
2345 if self.model is None:
2346 raise ValidationError({'model': _('Model must be selected')})
2348 if self.key is None:
2349 raise ValidationError({'key': _('Key must be selected')})
2351 if self.logical_key is None:
2352 raise ValidationError({'logical_key': _('Logical key must be selected')})
2354 # Ensure that the key is not the same as the logical key
2355 if self.key == self.logical_key:
2356 raise ValidationError({'key': _('Key must be different from logical key')})
2358 # Check against the reference status class
2359 status_class = self.get_status_class()
2361 if not status_class:
2362 raise ValidationError({
2363 'reference_status': _('Valid reference status class must be provided')
2364 })
2366 if self.key in status_class.values():
2367 raise ValidationError({
2368 'key': _(
2369 'Key must be different from the logical keys of the reference status'
2370 )
2371 })
2373 if self.logical_key not in status_class.values():
2374 raise ValidationError({
2375 'logical_key': _(
2376 'Logical key must be in the logical keys of the reference status'
2377 )
2378 })
2380 if self.name in status_class.names():
2381 raise ValidationError({
2382 'name': _(
2383 'Name must be different from the names of the reference status'
2384 )
2385 })
2387 return super().clean()
2389 def get_status_class(self):
2390 """Return the appropriate status class for this custom state."""
2391 from generic.states import StatusCode
2392 from InvenTree.helpers import inheritors
2394 if not self.reference_status:
2395 return None
2397 # Return the first class that matches the reference status
2398 for cls in inheritors(StatusCode):
2399 if cls.__name__ == self.reference_status:
2400 return cls
2403class SelectionList(InvenTree.models.MetadataMixin, InvenTree.models.InvenTreeModel):
2404 """Class which represents a list of selectable items for parameters.
2406 A lists selection options can be either manually defined, or sourced from a plugin.
2408 Attributes:
2409 name: The name of the selection list
2410 description: A description of the selection list
2411 locked: Is this selection list locked (i.e. cannot be modified)?
2412 active: Is this selection list active?
2413 source_plugin: The plugin which provides the selection list
2414 source_string: The string representation of the selection list
2415 default: The default value for the selection list
2416 created: The date/time that the selection list was created
2417 last_updated: The date/time that the selection list was last updated
2418 """
2420 class Meta:
2421 """Meta options for SelectionList."""
2423 verbose_name = _('Selection List')
2424 verbose_name_plural = _('Selection Lists')
2426 name = models.CharField(
2427 max_length=100,
2428 verbose_name=_('Name'),
2429 help_text=_('Name of the selection list'),
2430 unique=True,
2431 )
2433 description = models.CharField(
2434 max_length=250,
2435 verbose_name=_('Description'),
2436 help_text=_('Description of the selection list'),
2437 blank=True,
2438 )
2440 locked = models.BooleanField(
2441 default=False,
2442 verbose_name=_('Locked'),
2443 help_text=_('Is this selection list locked?'),
2444 )
2446 active = models.BooleanField(
2447 default=True,
2448 verbose_name=_('Active'),
2449 help_text=_('Can this selection list be used?'),
2450 )
2452 source_plugin = models.ForeignKey(
2453 'plugin.PluginConfig',
2454 on_delete=models.SET_NULL,
2455 blank=True,
2456 null=True,
2457 verbose_name=_('Source Plugin'),
2458 help_text=_('Plugin which provides the selection list'),
2459 )
2461 source_string = models.CharField(
2462 max_length=1000,
2463 verbose_name=_('Source String'),
2464 help_text=_('Optional string identifying the source used for this list'),
2465 blank=True,
2466 )
2468 default = models.ForeignKey(
2469 'SelectionListEntry',
2470 on_delete=models.SET_NULL,
2471 blank=True,
2472 null=True,
2473 verbose_name=_('Default Entry'),
2474 help_text=_('Default entry for this selection list'),
2475 )
2477 created = models.DateTimeField(
2478 auto_now_add=True,
2479 verbose_name=_('Created'),
2480 help_text=_('Date and time that the selection list was created'),
2481 )
2483 last_updated = models.DateTimeField(
2484 auto_now=True,
2485 verbose_name=_('Last Updated'),
2486 help_text=_('Date and time that the selection list was last updated'),
2487 )
2489 def __str__(self):
2490 """Return string representation of the selection list."""
2491 if not self.active:
2492 return f'{self.name} (Inactive)'
2493 return self.name
2495 @staticmethod
2496 def get_api_url():
2497 """Return the API URL associated with the SelectionList model."""
2498 return reverse('api-selectionlist-list')
2500 def get_choices(self, active: Optional[bool] = True):
2501 """Return the choices for the selection list.
2503 Arguments:
2504 active: If specified, filter choices by active status
2506 Returns:
2507 List of choice values for this selection list
2508 """
2509 choices = self.entries.all()
2511 if active is not None:
2512 choices = choices.filter(active=active)
2514 return [c.value for c in choices]
2516 def has_choice(self, value: str, active: Optional[bool] = None):
2517 """Check if the selection list has a particular choice.
2519 Arguments:
2520 value: The value to check for
2521 active: If specified, filter choices by active status
2523 Returns:
2524 True if the choice exists in the selection list, False otherwise
2525 """
2526 choices = self.entries.all()
2528 if active is not None:
2529 choices = choices.filter(active=active)
2531 return choices.filter(value=value).exists()
2534class SelectionListEntry(models.Model):
2535 """Class which represents a single entry in a SelectionList.
2537 Attributes:
2538 list: The SelectionList to which this entry belongs
2539 value: The value of the selection list entry
2540 label: The label for the selection list entry
2541 description: A description of the selection list entry
2542 active: Is this selection list entry active?
2543 """
2545 class Meta:
2546 """Meta options for SelectionListEntry."""
2548 verbose_name = _('Selection List Entry')
2549 verbose_name_plural = _('Selection List Entries')
2550 unique_together = [['list', 'value']]
2552 list = models.ForeignKey(
2553 SelectionList,
2554 on_delete=models.CASCADE,
2555 null=True,
2556 blank=True,
2557 related_name='entries',
2558 verbose_name=_('Selection List'),
2559 help_text=_('Selection list to which this entry belongs'),
2560 )
2562 value = models.CharField(
2563 max_length=255,
2564 verbose_name=_('Value'),
2565 help_text=_('Value of the selection list entry'),
2566 )
2568 label = models.CharField(
2569 max_length=255,
2570 verbose_name=_('Label'),
2571 help_text=_('Label for the selection list entry'),
2572 )
2574 description = models.CharField(
2575 max_length=250,
2576 verbose_name=_('Description'),
2577 help_text=_('Description of the selection list entry'),
2578 blank=True,
2579 )
2581 active = models.BooleanField(
2582 default=True,
2583 verbose_name=_('Active'),
2584 help_text=_('Is this selection list entry active?'),
2585 )
2587 def __str__(self):
2588 """Return string representation of the selection list entry."""
2589 if not self.active:
2590 return f'{self.label} (Inactive)'
2591 return self.label
2594class ParameterTemplate(
2595 InvenTree.models.MetadataMixin, InvenTree.models.InvenTreeModel
2596):
2597 """A ParameterTemplate provides a template for defining parameter values against various models.
2599 This allow for assigning arbitrary data fields against existing models,
2600 extending their functionality beyond the built-in fields.
2602 Attributes:
2603 name: The name (key) of the template
2604 description: A description of the template
2605 model_type: The type of model to which this template applies (e.g. 'part')
2606 units: The units associated with the template (if applicable)
2607 checkbox: Is this template a checkbox (boolean) type?
2608 choices: Comma-separated list of choices (if applicable)
2609 selectionlist: Optional link to a SelectionList for this template
2610 enabled: Is this template enabled?
2611 """
2613 IMPORT_ID_FIELDS = ['name']
2615 class Meta:
2616 """Metaclass options for the ParameterTemplate model."""
2618 verbose_name = _('Parameter Template')
2619 verbose_name_plural = _('Parameter Templates')
2621 # Note: Data was migrated from the existing 'part_partparametertemplate' table
2622 # Ref: https://github.com/inventree/InvenTree/pull/10699
2623 # To avoid data loss, we retain the existing table name
2624 db_table = 'part_partparametertemplate'
2626 class ModelChoices(RenderChoices):
2627 """Model choices for parameters."""
2629 choice_fnc = common.validators.parameter_template_model_options
2631 @staticmethod
2632 def get_api_url() -> str:
2633 """Return the API URL associated with the ParameterTemplate model."""
2634 return reverse('api-parameter-template-list')
2636 def __str__(self):
2637 """Return a string representation of a ParameterTemplate instance."""
2638 s = str(self.name)
2639 if self.units:
2640 s += f' ({self.units})'
2641 return s
2643 def clean(self):
2644 """Custom cleaning step for this model.
2646 Checks:
2647 - A 'checkbox' field cannot have 'choices' set
2648 - A 'checkbox' field cannot have 'units' set
2649 """
2650 super().clean()
2652 # Check that checkbox parameters do not have units or choices
2653 if self.checkbox: 2653 ↛ 2654line 2653 didn't jump to line 2654 because the condition on line 2653 was never true
2654 if self.units:
2655 raise ValidationError({
2656 'units': _('Checkbox parameters cannot have units')
2657 })
2659 if self.choices:
2660 raise ValidationError({
2661 'choices': _('Checkbox parameters cannot have choices')
2662 })
2664 # Check that 'choices' are in fact valid
2665 if self.choices is None: 2665 ↛ 2666line 2665 didn't jump to line 2666 because the condition on line 2665 was never true
2666 self.choices = ''
2667 else:
2668 self.choices = str(self.choices).strip()
2670 if self.choices: 2670 ↛ 2671line 2670 didn't jump to line 2671 because the condition on line 2670 was never true
2671 choice_set = set()
2673 for choice in self.choices.split(','):
2674 choice = choice.strip()
2676 # Ignore empty choices
2677 if not choice:
2678 continue
2680 if choice in choice_set:
2681 raise ValidationError({'choices': _('Choices must be unique')})
2683 choice_set.add(choice)
2685 def validate_unique(self, exclude=None):
2686 """Ensure that ParameterTemplates cannot be created with the same name.
2688 This test should be case-insensitive (which the unique caveat does not cover).
2689 """
2690 super().validate_unique(exclude)
2692 try:
2693 others = ParameterTemplate.objects.filter(name__iexact=self.name).exclude(
2694 pk=self.pk
2695 )
2697 if others.exists(): 2697 ↛ 2698line 2697 didn't jump to line 2698 because the condition on line 2697 was never true
2698 msg = _('Parameter template name must be unique')
2699 raise ValidationError({'name': msg})
2700 except ParameterTemplate.DoesNotExist:
2701 pass
2703 def get_choices(self):
2704 """Return a list of choices for this parameter template."""
2705 if self.selectionlist:
2706 return self.selectionlist.get_choices()
2708 if not self.choices:
2709 return []
2711 return [x.strip() for x in self.choices.split(',') if x.strip()]
2713 # TODO: Reintroduce validator for model_type
2714 model_type = models.ForeignKey(
2715 ContentType,
2716 on_delete=models.SET_NULL,
2717 blank=True,
2718 null=True,
2719 verbose_name=_('Model type'),
2720 help_text=_('Target model type for this parameter template'),
2721 )
2723 name = models.CharField(
2724 max_length=100,
2725 verbose_name=_('Name'),
2726 help_text=_('Parameter Name'),
2727 unique=True,
2728 )
2730 units = models.CharField(
2731 max_length=25,
2732 verbose_name=_('Units'),
2733 help_text=_('Physical units for this parameter'),
2734 blank=True,
2735 validators=[InvenTree.validators.validate_physical_units],
2736 )
2738 description = models.CharField(
2739 max_length=250,
2740 verbose_name=_('Description'),
2741 help_text=_('Parameter description'),
2742 blank=True,
2743 )
2745 checkbox = models.BooleanField(
2746 default=False,
2747 verbose_name=_('Checkbox'),
2748 help_text=_('Is this parameter a checkbox?'),
2749 )
2751 choices = models.CharField(
2752 max_length=5000,
2753 verbose_name=_('Choices'),
2754 help_text=_('Valid choices for this parameter (comma-separated)'),
2755 blank=True,
2756 )
2758 selectionlist = models.ForeignKey(
2759 SelectionList,
2760 blank=True,
2761 null=True,
2762 on_delete=models.SET_NULL,
2763 related_name='templates',
2764 verbose_name=_('Selection List'),
2765 help_text=_('Selection list for this parameter'),
2766 )
2768 enabled = models.BooleanField(
2769 default=True,
2770 verbose_name=_('Enabled'),
2771 help_text=_('Is this parameter template enabled?'),
2772 )
2775@receiver(
2776 post_save, sender=ParameterTemplate, dispatch_uid='post_save_parameter_template'
2777)
2778def post_save_parameter_template(sender, instance, created, **kwargs):
2779 """Callback function when a ParameterTemplate is created or saved."""
2780 import common.tasks
2782 if InvenTree.ready.canAppAccessDatabase() and not InvenTree.ready.isImportingData(): 2782 ↛ exitline 2782 didn't return from function 'post_save_parameter_template' because the condition on line 2782 was always true
2783 if not created:
2784 # Schedule a background task to rebuild the parameters against this template
2785 InvenTree.tasks.offload_task(
2786 common.tasks.rebuild_parameters,
2787 instance.pk,
2788 force_async=True,
2789 group='parameters',
2790 )
2793class Parameter(
2794 UpdatedUserMixin, InvenTree.models.MetadataMixin, InvenTree.models.InvenTreeModel
2795):
2796 """Class which represents a parameter value assigned to a particular model instance.
2798 Attributes:
2799 model_type: The type of model to which this parameter is linked
2800 model_id: The ID of the model to which this parameter is linked
2801 template: The ParameterTemplate which defines this parameter
2802 data: The value of the parameter [string]
2803 data_numeric: Numeric value of the parameter (if applicable) [float]
2804 note: Optional note associated with this parameter [string]
2805 updated: Date/time that this parameter was last updated
2806 updated_by: User who last updated this parameter
2807 """
2809 class Meta:
2810 """Meta options for Parameter model."""
2812 verbose_name = _('Parameter')
2813 verbose_name_plural = _('Parameters')
2814 unique_together = [['model_type', 'model_id', 'template']]
2815 indexes = [models.Index(fields=['model_type', 'model_id'])]
2817 # Note: Data was migrated from the existing 'part_partparameter' table
2818 # Ref: https://github.com/inventree/InvenTree/pull/10699
2819 # To avoid data loss, we retain the existing table name
2820 db_table = 'part_partparameter'
2822 class ModelChoices(RenderChoices):
2823 """Model choices for parameters."""
2825 choice_fnc = common.validators.parameter_model_options
2827 @staticmethod
2828 def get_api_url() -> str:
2829 """Return the API URL associated with the Parameter model."""
2830 return reverse('api-parameter-list')
2832 def save(self, *args, **kwargs):
2833 """Custom save method for Parameter model.
2835 - Update the numeric data field (if applicable)
2836 """
2837 self.calculate_numeric_value()
2839 # Convert 'boolean' values to 'True' / 'False'
2840 if self.template.checkbox:
2841 self.data = InvenTree.helpers.str2bool(self.data)
2842 self.data_numeric = 1 if self.data else 0
2844 self.check_save()
2845 super().save(*args, **kwargs)
2847 def delete(self):
2848 """Perform custom delete checks before deleting a Parameter instance."""
2849 self.check_delete()
2850 super().delete()
2852 def clean(self):
2853 """Validate the Parameter before saving to the database."""
2854 super().clean()
2856 # Validate the parameter data against the template choices
2857 if choices := self.template.get_choices():
2858 if self.data not in choices:
2859 raise ValidationError({'data': _('Invalid choice for parameter value')})
2861 self.calculate_numeric_value()
2863 # TODO: Check that the model_type for this parameter matches the template
2865 # Validate the parameter data against the template units
2866 if (
2867 get_global_setting(
2868 'PARAMETER_ENFORCE_UNITS', True, cache=False, create=False
2869 )
2870 and self.template.units
2871 ):
2872 try:
2873 InvenTree.conversion.convert_physical_value(
2874 self.data, self.template.units
2875 )
2876 except ValidationError as e:
2877 raise ValidationError({'data': e.message})
2879 if InvenTree.ready.isReadOnlyCommand():
2880 # Skip plugin validation checks during read-only management commands
2881 return
2883 # Finally, run custom validation checks (via plugins)
2884 from plugin import PluginMixinEnum, registry
2886 for plugin in registry.with_mixin(PluginMixinEnum.VALIDATION):
2887 # Note: The validate_parameter function may raise a ValidationError
2888 try:
2889 if hasattr(plugin, 'validate_parameter'):
2890 result = plugin.validate_parameter(self, self.data)
2891 if result:
2892 break
2893 except ValidationError as exc:
2894 # Re-throw the ValidationError against the 'data' field
2895 raise ValidationError({'data': exc.message})
2896 except Exception:
2897 InvenTree.exceptions.log_error('validate_parameter', plugin=plugin.slug)
2899 def calculate_numeric_value(self):
2900 """Calculate a numeric value for the parameter data.
2902 - If a 'units' field is provided, then the data will be converted to the base SI unit.
2903 - Otherwise, we'll try to do a simple float cast
2904 """
2905 if self.template.units:
2906 try:
2907 self.data_numeric = InvenTree.conversion.convert_physical_value(
2908 self.data, self.template.units
2909 )
2910 except (ValidationError, ValueError):
2911 self.data_numeric = None
2913 # No units provided, so try to cast to a float
2914 else:
2915 try:
2916 self.data_numeric = float(self.data)
2917 except ValueError:
2918 self.data_numeric = None
2920 if self.data_numeric is not None and type(self.data_numeric) is float:
2921 # Prevent out of range numbers, etc
2922 # Ref: https://github.com/inventree/InvenTree/issues/7593
2923 if math.isnan(self.data_numeric) or math.isinf(self.data_numeric):
2924 self.data_numeric = None
2926 def check_permission(self, permission, user):
2927 """Check if the user has the required permission for this parameter."""
2928 from InvenTree.models import InvenTreeParameterMixin
2930 model_class = self.model_type.model_class()
2932 if not issubclass(model_class, InvenTreeParameterMixin):
2933 raise ValidationError(_('Invalid model type specified for parameter'))
2935 return model_class.check_related_permission(permission, user)
2937 def check_save(self):
2938 """Check if this parameter can be saved.
2940 The linked content_object can implement custom checks by overriding
2941 the 'check_parameter_edit' method.
2942 """
2943 from InvenTree.models import InvenTreeParameterMixin
2945 try:
2946 instance = self.content_object
2947 except InvenTree.models.InvenTreeModel.DoesNotExist:
2948 return
2950 if instance and isinstance(instance, InvenTreeParameterMixin):
2951 instance.check_parameter_save(self)
2953 def check_delete(self):
2954 """Check if this parameter can be deleted."""
2955 from InvenTree.models import InvenTreeParameterMixin
2957 try:
2958 instance = self.content_object
2959 except InvenTree.models.InvenTreeModel.DoesNotExist:
2960 return
2962 if instance and isinstance(instance, InvenTreeParameterMixin):
2963 instance.check_parameter_delete(self)
2965 # TODO: Reintroduce validator for model_type
2966 model_type = models.ForeignKey(ContentType, on_delete=models.CASCADE)
2968 model_id = models.PositiveIntegerField(
2969 verbose_name=_('Model ID'),
2970 help_text=_('ID of the target model for this parameter'),
2971 )
2973 content_object = GenericForeignKey('model_type', 'model_id')
2975 template = models.ForeignKey(
2976 ParameterTemplate,
2977 on_delete=models.CASCADE,
2978 related_name='parameters',
2979 verbose_name=_('Template'),
2980 help_text=_('Parameter template'),
2981 )
2983 data = models.CharField(
2984 max_length=500,
2985 verbose_name=_('Data'),
2986 help_text=_('Parameter Value'),
2987 validators=[MinLengthValidator(1)],
2988 )
2990 data_numeric = models.FloatField(default=None, null=True, blank=True)
2992 note = models.CharField(
2993 max_length=500,
2994 blank=True,
2995 verbose_name=_('Note'),
2996 help_text=_('Optional note field'),
2997 )
2999 @property
3000 def units(self):
3001 """Return the units associated with the template."""
3002 return self.template.units
3004 @property
3005 def name(self):
3006 """Return the name of the template."""
3007 return self.template.name
3009 @property
3010 def description(self):
3011 """Return the description of the template."""
3012 return self.template.description
3015class BarcodeScanResult(InvenTree.models.InvenTreeModel):
3016 """Model for storing barcode scans results."""
3018 BARCODE_SCAN_MAX_LEN = 250
3020 class Meta:
3021 """Model meta options."""
3023 verbose_name = _('Barcode Scan')
3025 data = models.CharField(
3026 max_length=BARCODE_SCAN_MAX_LEN,
3027 verbose_name=_('Data'),
3028 help_text=_('Barcode data'),
3029 blank=False,
3030 null=False,
3031 )
3033 user = models.ForeignKey(
3034 User,
3035 on_delete=models.SET_NULL,
3036 blank=True,
3037 null=True,
3038 verbose_name=_('User'),
3039 help_text=_('User who scanned the barcode'),
3040 )
3042 timestamp = models.DateTimeField(
3043 auto_now_add=True,
3044 verbose_name=_('Timestamp'),
3045 help_text=_('Date and time of the barcode scan'),
3046 )
3048 endpoint = models.CharField(
3049 max_length=250,
3050 verbose_name=_('Path'),
3051 help_text=_('URL endpoint which processed the barcode'),
3052 blank=True,
3053 null=True,
3054 )
3056 context = models.JSONField(
3057 max_length=1000,
3058 verbose_name=_('Context'),
3059 help_text=_('Context data for the barcode scan'),
3060 blank=True,
3061 null=True,
3062 )
3064 response = models.JSONField(
3065 max_length=1000,
3066 verbose_name=_('Response'),
3067 help_text=_('Response data from the barcode scan'),
3068 blank=True,
3069 null=True,
3070 )
3072 result = models.BooleanField(
3073 verbose_name=_('Result'),
3074 help_text=_('Was the barcode scan successful?'),
3075 default=False,
3076 )
3079class DataOutput(models.Model):
3080 """Model for storing generated data output from various processes.
3082 This model is intended for storing data files which are generated by various processes,
3083 and need to be retained for future use (e.g. download by the user).
3085 Attributes:
3086 created: Date and time that the data output was created
3087 user: User who created the data output (if applicable)
3088 total: Total number of items / records in the data output
3089 progress: Current progress of the data output generation process
3090 complete: Has the data output generation process completed?
3091 output_type: The type of data output generated (e.g. 'label', 'report', etc)
3092 template_name: Name of the template used to generate the data output (if applicable)
3093 plugin: Key for the plugin which generated the data output (if applicable)
3094 output: File field for storing the generated file
3095 errors: JSON field for storing any errors generated during the data output generation process
3096 """
3098 class DataOutputTypes(StringEnum):
3099 """Enum for data output types."""
3101 LABEL = 'label'
3102 REPORT = 'report'
3103 EXPORT = 'export'
3105 created = models.DateField(auto_now_add=True, editable=False)
3107 user = models.ForeignKey(
3108 User, on_delete=models.SET_NULL, blank=True, null=True, related_name='+'
3109 )
3111 total = models.PositiveIntegerField(default=1)
3113 progress = models.PositiveIntegerField(default=0)
3115 complete = models.BooleanField(default=False)
3117 output_type = models.CharField(max_length=100, blank=True, null=True)
3119 template_name = models.CharField(max_length=100, blank=True, null=True)
3121 plugin = models.CharField(max_length=100, blank=True, null=True)
3123 output = models.FileField(upload_to='data_output', blank=True, null=True)
3125 errors = models.JSONField(blank=True, null=True)
3127 def mark_complete(self, progress: int = 100, output: Optional[ContentFile] = None):
3128 """Mark the data output generation process as complete.
3130 Arguments:
3131 progress (int, optional): Progress percentage of the data output generation. Defaults to 100.
3132 output (ContentFile, optional): The generated output file. Defaults to None.
3133 """
3134 self.complete = True
3135 self.progress = progress
3136 self.output = output
3137 self.save()
3139 def mark_failure(
3140 self, error: Optional[str] = None, error_dict: Optional[dict] = None
3141 ):
3142 """Log an error message to the errors field.
3144 Arguments:
3145 error (str, optional): Error message to log. Defaults to None.
3146 error_dict (dict): Dictionary containing error messages. Defaults to None.
3147 """
3148 self.complete = False
3149 self.output = None
3151 if error_dict is not None:
3152 self.errors = error_dict
3153 elif error is not None:
3154 self.errors = {'error': str(error)}
3155 else:
3156 self.errors = {'error': str(_('An error occurred'))}
3158 self.save()
3161# region Email
3162class Priority(models.IntegerChoices):
3163 """Enumeration for defining email priority levels."""
3165 NONE = 0
3166 VERY_HIGH = 1
3167 HIGH = 2
3168 NORMAL = 3
3169 LOW = 4
3170 VERY_LOW = 5
3173HEADER_PRIORITY = 'X-Priority'
3174HEADER_MSG_ID = 'Message-ID'
3176del_error_msg = _(
3177 'INVE-E8: Email log deletion is protected. Set INVENTREE_PROTECT_EMAIL_LOG to False to allow deletion.'
3178)
3181class NoDeleteQuerySet(models.query.QuerySet):
3182 """Custom QuerySet to prevent deletion of EmailLog entries."""
3184 def delete(self):
3185 """Override delete method to prevent deletion of EmailLog entries."""
3186 if get_global_setting('INVENTREE_PROTECT_EMAIL_LOG'):
3187 raise ValidationError(del_error_msg)
3188 super().delete()
3191class NoDeleteManager(models.Manager):
3192 """Custom Manager to use NoDeleteQuerySet."""
3194 def get_queryset(self):
3195 """Return a NoDeleteQuerySet."""
3196 return NoDeleteQuerySet(self.model, using=self._db)
3199class EmailMessage(models.Model):
3200 """Model for storing email messages sent or received by the system.
3202 Attributes:
3203 global_id: Unique identifier for the email message
3204 message_id_key: Identifier for the email message - might be supplied by external system
3205 thread_id_key: Identifier of thread - might be supplied by external system
3206 subject: Subject of the email message
3207 body: Body of the email message
3208 to: Recipient of the email message
3209 sender: Sender of the email message
3210 status: Status of the email message (e.g. 'sent', 'failed', etc)
3211 timestamp: Date and time that the email message left the system or was received by the system
3212 headers: Headers of the email message
3213 full_message: Full email message content
3214 direction: Direction of the email message (e.g. 'inbound', 'outbound')
3215 error_code: Error code (if applicable)
3216 error_message: Error message (if applicable)
3217 error_timestamp: Date and time of the error (if applicable)
3218 delivery_options: Delivery options for the email message
3219 """
3221 class Meta:
3222 """Meta options for EmailMessage."""
3224 verbose_name = _('Email Message')
3225 verbose_name_plural = _('Email Messages')
3227 class EmailStatus(models.TextChoices):
3228 """Machine setting config type enum."""
3230 ANNOUNCED = (
3231 'A',
3232 _('Announced'),
3233 ) # Intend to send mail was announced (saved in system, pushed to queue)
3234 SENT = 'S', _('Sent') # Mail was sent to the email server
3235 FAILED = 'F', _('Failed') # There was en error sending the email
3236 DELIVERED = (
3237 'D',
3238 _('Delivered'),
3239 ) # Mail was delivered to the recipient - this means we got some kind of feedback from the email server or user
3240 READ = (
3241 'R',
3242 _('Read'),
3243 ) # Mail was read by the recipient - this means we got some kind of feedback from the user
3244 CONFIRMED = (
3245 'C',
3246 _('Confirmed'),
3247 ) # Mail delivery was confirmed by the recipient explicitly
3249 class EmailDirection(models.TextChoices):
3250 """Email direction enum."""
3252 INBOUND = 'I', _('Inbound')
3253 OUTBOUND = 'O', _('Outbound')
3255 class DeliveryOptions(models.TextChoices):
3256 """Email delivery options enum."""
3258 NO_REPLY = 'no_reply', _('No Reply')
3259 TRACK_DELIVERY = 'track_delivery', _('Track Delivery')
3260 TRACK_READ = 'track_read', _('Track Read')
3261 TRACK_CLICK = 'track_click', _('Track Click')
3263 global_id = models.UUIDField(
3264 verbose_name=_('Global ID'),
3265 help_text=_('Unique identifier for this message'),
3266 primary_key=True,
3267 default=uuid.uuid4,
3268 editable=False,
3269 unique=True,
3270 )
3271 message_id_key = models.CharField(
3272 max_length=250,
3273 blank=True,
3274 null=True,
3275 verbose_name=_('Message ID'),
3276 help_text=_(
3277 'Identifier for this message (might be supplied by external system)'
3278 ),
3279 )
3280 thread_id_key = models.CharField(
3281 max_length=250,
3282 blank=True,
3283 null=True,
3284 verbose_name=_('Thread ID'),
3285 help_text=_(
3286 'Identifier for this message thread (might be supplied by external system)'
3287 ),
3288 )
3289 thread = models.ForeignKey(
3290 'EmailThread',
3291 on_delete=models.SET_NULL,
3292 blank=True,
3293 null=True,
3294 related_name='messages',
3295 verbose_name=_('Thread'),
3296 help_text=_('Linked thread for this message'),
3297 )
3298 subject = models.CharField(max_length=250, blank=False, null=False)
3299 body = models.TextField(blank=False, null=False)
3300 to = models.EmailField(blank=False, null=False)
3301 sender = models.EmailField(blank=False, null=False)
3302 status = models.CharField(
3303 max_length=50, blank=True, null=True, choices=EmailStatus.choices
3304 )
3305 timestamp = models.DateTimeField(auto_now_add=True, editable=False)
3306 headers = models.JSONField(blank=True, null=True)
3307 # Additional info
3308 full_message = models.TextField(blank=True, null=True)
3309 direction = models.CharField(
3310 max_length=50, blank=True, null=True, choices=EmailDirection.choices
3311 )
3312 priority = models.IntegerField(verbose_name=_('Priority'), choices=Priority)
3313 delivery_options = models.JSONField(
3314 blank=True,
3315 null=True,
3316 # choices=DeliveryOptions.choices
3317 )
3318 # Optional tracking of delivery
3319 error_code = models.CharField(max_length=50, blank=True, null=True)
3320 error_message = models.TextField(blank=True, null=True)
3321 error_timestamp = models.DateTimeField(blank=True, null=True)
3323 def save(self, *args, **kwargs):
3324 """Ensure threads exist before saving the email message."""
3325 ret = super().save(*args, **kwargs)
3327 # Ensure thread is linked
3328 if not self.thread:
3329 thread, created = EmailThread.objects.get_or_create(
3330 key=self.thread_id_key, started_internal=True
3331 )
3332 self.thread = thread
3333 if created and not self.thread_id_key:
3334 self.thread_id_key = thread.global_id
3335 self.save()
3337 return ret
3339 objects = NoDeleteManager()
3341 def delete(self, *kwargs):
3342 """Delete entry - if not protected."""
3343 if get_global_setting('INVENTREE_PROTECT_EMAIL_LOG'):
3344 raise ValidationError(del_error_msg)
3345 return super().delete(*kwargs)
3348class EmailThread(InvenTree.models.InvenTreeMetadataModel):
3349 """Model for storing email threads."""
3351 class Meta:
3352 """Meta options for EmailThread."""
3354 verbose_name = _('Email Thread')
3355 verbose_name_plural = _('Email Threads')
3356 unique_together = [['key', 'global_id']]
3357 ordering = ['-updated']
3359 @staticmethod
3360 def get_api_url():
3361 """Return the API URL associated with the EmailThread model."""
3362 return reverse('api-email-list')
3364 key = models.CharField(
3365 max_length=250,
3366 verbose_name=_('Key'),
3367 null=True,
3368 blank=True,
3369 help_text=_('Unique key for this thread (used to identify the thread)'),
3370 )
3371 global_id = models.UUIDField(
3372 verbose_name=_('Global ID'),
3373 help_text=_('Unique identifier for this thread'),
3374 primary_key=True,
3375 default=uuid.uuid4,
3376 editable=False,
3377 )
3378 started_internal = models.BooleanField(
3379 default=False,
3380 verbose_name=_('Started Internal'),
3381 help_text=_('Was this thread started internally?'),
3382 )
3383 created = models.DateTimeField(
3384 auto_now_add=True,
3385 verbose_name=_('Created'),
3386 help_text=_('Date and time that the thread was created'),
3387 )
3388 updated = models.DateTimeField(
3389 auto_now=True,
3390 verbose_name=_('Updated'),
3391 help_text=_('Date and time that the thread was last updated'),
3392 )
3395def issue_mail(
3396 subject: str,
3397 body: str,
3398 from_email: str,
3399 recipients: str | list,
3400 fail_silently: bool = False,
3401 html_message=None,
3402 prio: Priority = Priority.NORMAL,
3403 headers: Optional[dict] = None,
3404):
3405 """Send an email with the specified subject and body, to the specified recipients list.
3407 Mostly used by tasks.
3408 """
3409 connection = get_connection(fail_silently=fail_silently)
3411 message = EmailMultiAlternatives(
3412 subject, body, from_email, recipients, connection=connection
3413 )
3414 if html_message:
3415 message.attach_alternative(html_message, 'text/html')
3417 # Add any extra headers
3418 if headers is not None:
3419 for key, value in headers.items():
3420 message.extra_headers[key] = value
3422 # Stabilize the message ID before creating the object
3423 if HEADER_MSG_ID not in message.extra_headers:
3424 message.extra_headers[HEADER_MSG_ID] = make_msgid(domain=DNS_NAME)
3426 # TODO add `References` field for the thread ID
3428 # Add headers for flags
3429 message.extra_headers[HEADER_PRIORITY] = str(prio)
3431 # And now send
3432 return message.send()
3435def log_email_messages(email_messages) -> list[EmailMessage]:
3436 """Log email messages to the database.
3438 Args:
3439 email_messages (list): List of email messages to log.
3440 """
3441 instance_id = inventree_identifier(True)
3443 msg_ids = []
3444 for msg in email_messages:
3445 try:
3446 new_obj = EmailMessage.objects.create(
3447 message_id_key=msg.extra_headers.get(HEADER_MSG_ID),
3448 subject=msg.subject,
3449 body=msg.body,
3450 to=msg.to,
3451 sender=msg.from_email,
3452 status=EmailMessage.EmailStatus.ANNOUNCED,
3453 direction=EmailMessage.EmailDirection.OUTBOUND,
3454 priority=msg.extra_headers.get(HEADER_PRIORITY, '3'),
3455 headers=msg.extra_headers,
3456 full_message=msg,
3457 )
3458 msg_ids.append(new_obj)
3460 # Add InvenTree specific headers to the message to help with identification if we see mails again
3461 msg.extra_headers['X-InvenTree-MsgId-1'] = str(new_obj.global_id)
3462 msg.extra_headers['X-InvenTree-ThreadId-1'] = str(new_obj.thread.global_id)
3463 msg.extra_headers['X-InvenTree-Instance-1'] = str(instance_id)
3464 except Exception as exc: # pragma: no cover
3465 logger.error(f' INVE-W10: Failed to log email message: {exc}')
3466 return msg_ids
3469@receiver(inbound)
3470def handle_inbound(sender, event, esp_name, **kwargs):
3471 """Handle inbound email messages from anymail."""
3472 message = event.message
3474 r_to = message.envelope_recipient or [a.addr_spec for a in message.to]
3475 r_sender = message.envelope_sender or message.from_email.addr_spec
3477 msg = EmailMessage.objects.create(
3478 message_id_key=event.message[HEADER_MSG_ID],
3479 subject=message.subject,
3480 body=message.text,
3481 to=r_to,
3482 sender=r_sender,
3483 status=EmailMessage.EmailStatus.READ,
3484 direction=EmailMessage.EmailDirection.INBOUND,
3485 priority=Priority.NONE,
3486 timestamp=message.date,
3487 headers=message._headers,
3488 full_message=message.html,
3489 )
3491 # Schedule a task to process the email message
3492 from plugin.base.mail.mail import process_mail_in
3494 InvenTree.tasks.offload_task(process_mail_in, mail_id=msg.pk, group='mail')
3497@receiver(tracking)
3498def handle_event(sender, event, esp_name, **kwargs):
3499 """Handle tracking events from anymail."""
3500 try:
3501 email = EmailMessage.objects.get(message_id_key=event.message_id)
3503 if event.event_type == 'delivered':
3504 email.status = EmailMessage.EmailStatus.DELIVERED
3505 elif event.event_type == 'opened':
3506 email.status = EmailMessage.EmailStatus.READ
3507 elif event.event_type == 'clicked':
3508 email.status = EmailMessage.EmailStatus.CONFIRMED
3509 elif event.event_type == 'sent':
3510 email.status = EmailMessage.EmailStatus.SENT
3511 elif event.event_type == 'unknown':
3512 email.error_message = event.esp_event
3513 else:
3514 if event.event_type in ('queued', 'deferred'):
3515 # We ignore these
3516 return True
3517 else:
3518 email.status = EmailMessage.EmailStatus.FAILED
3519 email.error_code = event.event_type
3520 email.error_message = event.esp_event
3521 email.error_timestamp = event.timestamp
3522 email.save()
3523 return True
3524 except EmailMessage.DoesNotExist:
3525 return False
3526 except Exception as exc: # pragma: no cover
3527 logger.error(f' INVE-W10: Failed to handle tracking event: {exc}')
3528 return False
3531# endregion Email
3533# region tracing for django q
3534if TRACE_PROC: # pragma: no cover 3534 ↛ 3536line 3534 didn't jump to line 3536 because the condition on line 3534 was never true
3536 @receiver(post_spawn)
3537 def spawn_callback(sender, proc_name, **kwargs):
3538 """Callback to patch in tracing support."""
3539 TRACE_PROV.add_span_processor(TRACE_PROC)
3540 trace.set_tracer_provider(TRACE_PROV)
3541 trace.get_tracer(__name__)
3543# endregion