Coverage for src/backend/InvenTree/generic/states/fields.py: 71%
148 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 17:47 +0000
1"""Custom model/serializer fields for InvenTree models that support custom states."""
3from collections.abc import Iterable
4from typing import Any, Optional
6from django.core.exceptions import ObjectDoesNotExist
7from django.db import models
8from django.utils.encoding import force_str
9from django.utils.translation import gettext_lazy as _
11from drf_spectacular.types import OpenApiTypes
12from drf_spectacular.utils import extend_schema_field
13from rest_framework import serializers
14from rest_framework.fields import ChoiceField
16import InvenTree.ready
18from .custom import get_logical_value
21class CustomChoiceField(serializers.ChoiceField):
22 """Custom Choice Field.
24 This is not intended to be used directly.
25 """
27 def __init__(self, choices: Iterable, **kwargs):
28 """Initialize the field."""
29 choice_mdl = kwargs.pop('choice_mdl', None)
30 choice_field = kwargs.pop('choice_field', None)
31 is_custom = kwargs.pop('is_custom', False)
32 kwargs.pop('max_value', None)
33 kwargs.pop('min_value', None)
34 super().__init__(choices, **kwargs)
35 self.choice_mdl = choice_mdl
36 self.choice_field = choice_field
37 self.is_custom = is_custom
39 def to_internal_value(self, data):
40 """Map the choice (that might be a custom one) back to the logical value."""
41 try:
42 return super().to_internal_value(data)
43 except serializers.ValidationError:
44 try:
45 logical = get_logical_value(data, self.choice_mdl._meta.model_name)
46 if self.is_custom:
47 return logical.key
48 return logical.logical_key
49 except (ObjectDoesNotExist, Exception):
50 raise serializers.ValidationError('Invalid choice')
52 def get_field_info(self, field, field_info):
53 """Return the field information for the given item."""
54 from common.models import InvenTreeCustomUserStateModel
56 # Static choices
57 choices = [
58 {
59 'value': choice_value,
60 'display_name': force_str(choice_name, strings_only=True),
61 }
62 for choice_value, choice_name in field.choices.items()
63 ]
64 # Dynamic choices from InvenTreeCustomUserStateModel
65 objs = InvenTreeCustomUserStateModel.objects.filter(
66 model__model=field.choice_mdl._meta.model_name
67 )
68 dyn_choices = [
69 {'value': choice.key, 'display_name': choice.label} for choice in objs.all()
70 ]
72 if dyn_choices: 72 ↛ 73line 72 didn't jump to line 73 because the condition on line 72 was never true
73 all_choices = choices + dyn_choices
74 field_info['choices'] = sorted(all_choices, key=lambda kv: kv['value'])
75 else:
76 field_info['choices'] = choices
77 return field_info
80@extend_schema_field(OpenApiTypes.INT)
81class ExtraCustomChoiceField(CustomChoiceField):
82 """Custom Choice Field that returns value of status if empty.
84 This is not intended to be used directly.
85 """
87 def to_representation(self, value):
88 """Return the value of the status if it is empty."""
89 return super().to_representation(value) or value
92class InvenTreeCustomStatusModelField(models.PositiveIntegerField):
93 """Custom model field for extendable status codes.
95 Adds a secondary *_custom_key field to the model which can be used to store additional status information.
96 Models using this model field must also include the InvenTreeCustomStatusSerializerMixin in all serializers that create or update the value.
97 """
99 def __init__(self, *args, **kwargs):
100 """Initialize the field."""
101 from generic.states.validators import CustomStatusCodeValidator
103 self.status_class = kwargs.pop('status_class', None)
105 validators = kwargs.pop('validators', None) or []
107 if self.status_class:
108 validators.append(CustomStatusCodeValidator(status_class=self.status_class))
110 kwargs['validators'] = validators
111 super().__init__(*args, **kwargs)
113 def deconstruct(self):
114 """Deconstruct the field for migrations."""
115 name, path, args, kwargs = super().deconstruct()
117 return name, path, args, kwargs
119 def contribute_to_class(self, cls, name):
120 """Add the _custom_key field to the model."""
121 cls._meta.supports_custom_status = True
123 if not hasattr(self, '_custom_key_field') and not hasattr(
124 cls, f'{name}_custom_key'
125 ):
126 self.add_field(cls, name)
128 super().contribute_to_class(cls, name)
130 def clean(self, value: Any, model_instance: Any) -> Any:
131 """Ensure that the value is not an empty string."""
132 if value == '': 132 ↛ 133line 132 didn't jump to line 133 because the condition on line 132 was never true
133 value = None
135 return super().clean(value, model_instance)
137 def add_field(self, cls, name):
138 """Adds custom_key_field to the model class to save additional status information."""
139 from generic.states.validators import CustomStatusCodeValidator
141 validators = []
143 if self.status_class:
144 validators.append(CustomStatusCodeValidator(status_class=self.status_class))
146 help_text = _('Additional status information for this item')
147 if InvenTree.ready.isGeneratingSchema() and self.status_class: 147 ↛ 148line 147 didn't jump to line 148 because the condition on line 147 was never true
148 help_text = (
149 help_text
150 + '\n\n'
151 + '\n'.join(
152 f'* `{value}` - {label}'
153 for value, label in self.status_class.items(custom=True)
154 )
155 + "\n\nAdditional custom status keys may be retrieved from the corresponding 'status_retrieve' call."
156 )
158 custom_key_field = ExtraInvenTreeCustomStatusModelField(
159 default=None,
160 verbose_name=_('Custom status key'),
161 help_text=help_text,
162 validators=validators,
163 blank=True,
164 null=True,
165 )
167 cls.add_to_class(f'{name}_custom_key', custom_key_field)
168 self._custom_key_field = custom_key_field
171class ExtraInvenTreeCustomStatusModelField(models.PositiveIntegerField):
172 """Custom field used to detect custom extenteded fields.
174 This is not intended to be used directly, if you want to support custom states in your model use InvenTreeCustomStatusModelField.
175 """
177 def __init__(self, *args, **kwargs):
178 """Initialize the field."""
179 super().__init__(*args, **kwargs)
182class InvenTreeCustomStatusSerializerMixin:
183 """Mixin to ensure custom status fields are set.
185 This mixin must be used to ensure that custom status fields are set correctly when updating a model.
186 """
188 _custom_fields: Optional[list] = None
189 _custom_fields_leader: Optional[list] = None
190 _custom_fields_follower: Optional[list] = None
191 _is_gathering = False
193 def update(self, instance, validated_data):
194 """Ensure the custom field is updated if the leader was changed."""
195 self.gather_custom_fields()
196 # Mirror values from leader to follower
197 for field in self._custom_fields_leader or []: 197 ↛ 198line 197 didn't jump to line 198 because the loop on line 197 never started
198 follower_field_name = f'{field}_custom_key'
199 if (
200 field in self.initial_data
201 and self.instance
202 and self.initial_data[field]
203 != getattr(self.instance, follower_field_name, None)
204 ):
205 setattr(self.instance, follower_field_name, self.initial_data[field])
207 # Mirror values from follower to leader
208 for field in self._custom_fields_follower or []: 208 ↛ 209line 208 didn't jump to line 209 because the loop on line 208 never started
209 leader_field_name = field.replace('_custom_key', '')
210 if field in validated_data and leader_field_name not in self.initial_data:
211 try:
212 reference = get_logical_value(
213 validated_data[field],
214 self.fields[field].choice_mdl._meta.model_name,
215 )
216 validated_data[leader_field_name] = reference.logical_key
217 except (ObjectDoesNotExist, Exception):
218 if validated_data[field] in self.fields[leader_field_name].choices:
219 validated_data[leader_field_name] = validated_data[field]
220 else:
221 raise serializers.ValidationError('Invalid choice')
222 return super().update(instance, validated_data)
224 def to_representation(self, instance):
225 """Ensure custom state fields are not served empty."""
226 data = super().to_representation(instance)
227 for field in self.gather_custom_fields(): 227 ↛ 228line 227 didn't jump to line 228 because the loop on line 227 never started
228 if data[field] is None:
229 data[field] = data[
230 field.replace('_custom_key', '')
231 ] # Use "normal" status field instead
232 return data
234 def gather_custom_fields(self):
235 """Gather all custom fields on the serializer."""
236 if self._custom_fields_follower: 236 ↛ 237line 236 didn't jump to line 237 because the condition on line 236 was never true
237 self._is_gathering = False
238 return self._custom_fields_follower
240 if self._is_gathering:
241 self._custom_fields = {}
242 else:
243 self._is_gathering = True
244 # Gather fields
245 self._custom_fields = {
246 k: v.is_custom
247 for k, v in self.fields.items()
248 if isinstance(v, CustomChoiceField)
249 }
251 # Separate fields for easier/cheaper access
252 self._custom_fields_follower = [k for k, v in self._custom_fields.items() if v]
253 self._custom_fields_leader = [
254 k for k, v in self._custom_fields.items() if not v
255 ]
257 return self._custom_fields_follower
259 def build_standard_field(self, field_name, model_field):
260 """Use custom field for custom status model.
262 This is required because of DRF overwriting all fields with choice sets.
263 """
264 field_cls, field_kwargs = super().build_standard_field(field_name, model_field)
265 if issubclass(field_cls, ChoiceField) and isinstance(
266 model_field, InvenTreeCustomStatusModelField
267 ):
268 field_cls = CustomChoiceField
269 field_kwargs['choice_mdl'] = model_field.model
270 field_kwargs['choice_field'] = model_field.name
271 elif isinstance(model_field, ExtraInvenTreeCustomStatusModelField):
272 field_cls = ExtraCustomChoiceField
273 field_kwargs['choice_mdl'] = model_field.model
274 field_kwargs['choice_field'] = model_field.name
275 field_kwargs['is_custom'] = True
277 # Inherit choices from leader
278 self.gather_custom_fields()
279 if self._custom_fields and field_name in self._custom_fields:
280 leader_field_name = field_name.replace('_custom_key', '')
281 leader_field = self.fields[leader_field_name]
282 if hasattr(leader_field, 'choices'): 282 ↛ 284line 282 didn't jump to line 284 because the condition on line 282 was always true
283 field_kwargs['choices'] = list(leader_field.choices.items())
284 elif hasattr(model_field.model, leader_field_name):
285 leader_model_field = getattr(
286 model_field.model, leader_field_name
287 ).field
288 if hasattr(leader_model_field, 'choices'):
289 field_kwargs['choices'] = leader_model_field.choices
291 if getattr(leader_field, 'read_only', False) is True:
292 field_kwargs['read_only'] = True
294 if 'choices' not in field_kwargs:
295 field_kwargs['choices'] = []
297 return field_cls, field_kwargs