Coverage for src/backend/InvenTree/InvenTree/serializers.py: 76%

1"""Serializers used in various InvenTree apps.""" 

2 

3from collections import OrderedDict 

4from copy import deepcopy 

5from dataclasses import dataclass 

6from decimal import Decimal 

7from typing import Optional 

8 

9from django.contrib.contenttypes.models import ContentType 

10from django.core.exceptions import ValidationError as DjangoValidationError 

11from django.core.files.storage import default_storage 

12from django.db import models 

13from django.db.models import QuerySet 

14from django.utils.translation import gettext_lazy as _ 

15 

16from djmoney.contrib.django_rest_framework.fields import MoneyField 

17from djmoney.money import Money 

18from djmoney.utils import MONEY_CLASSES, get_currency_field_name 

19from drf_spectacular.utils import extend_schema_field 

20from rest_framework import serializers 

21from rest_framework.exceptions import ValidationError 

22from rest_framework.fields import empty 

23from rest_framework.mixins import ListModelMixin 

24from rest_framework.permissions import SAFE_METHODS 

25from rest_framework.serializers import DecimalField, Serializer 

26from rest_framework.utils import model_meta 

27from taggit.serializers import TaggitSerializer 

28 

29import InvenTree.ready 

30from common.currency import currency_code_default, currency_code_mappings 

31from InvenTree.fields import InvenTreeRestURLField, InvenTreeURLField 

32from InvenTree.helpers import str2bool 

33from InvenTree.helpers_model import getModelsWithMixin 

34 

35 

36@dataclass 

37class OptionalField: 

38 """DataClass used to optionally enable a serializer field. 

39 

40 This is used in conjunction with the `FilterableSerializerMixin` to allow 

41 dynamic inclusion or exclusion of serializer fields at runtime. 

42 

43 Adding OptionalField instances to a serializer class is more "efficient" 

44 than directly adding the field (and later removing it), 

45 as the field is never instantiated unless it is required. 

46 

47 Additionally, you can specify prefetch fields which will be applied 

48 to the queryset, *only* if the field is included in the final serializer. 

49 

50 This allows for optimization of database queries based only on the requested data. 

51 

52 Example: 

53 class MySerializer(FilterableSerializerMixin, serializers.ModelSerializer): 

54 my_optional_field = OptionalField( 

55 serializer_class=serializers.CharField, 

56 default_include=False, 

57 filter_name='include_my_field', 

58 serializer_kwargs={ 

59 'help_text': 'This is an optional field', 

60 'read_only': True, 

61 }, 

62 prefetch_fields=['related_field'], 

63 ) 

64 

65 """ 

66 

67 serializer_class: Serializer 

68 serializer_kwargs: Optional[dict] = None 

69 default_include: bool = False 

70 filter_name: Optional[str] = None 

71 filter_by_query: bool = True 

72 prefetch_fields: Optional[list[str]] = None 

73 

74 

75class FilterableSerializerMixin: 

76 """Mixin that enables filtering of marked fields on a serializer. 

77 

78 Use the `OptionalField` helper class to mark serializer fields as filterable. 

79 This introduces overhead during initialization, so only use this mixin when necessary. 

80 """ 

81 

82 optional_filters: dict = None 

83 fields_to_remove: set = None 

84 optional_fields: set = None 

85 filter_on_query: bool = True 

86 

87 def __init__(self, *args, **kwargs): 

88 """Initialization routine for the serializer. This gathers and applies filters through kwargs.""" 

89 # Extract some useful context information for later use 

90 context = kwargs.get('context', {}) 

91 self.request = context.get('request', None) or getattr(self, 'request', None) 

92 self.request_query_params = ( 

93 dict(getattr(self.request, 'query_params', {})) if self.request else {} 

94 ) 

95 

96 self.gather_optional_fields(kwargs) 

97 

98 super().__init__(*args, **kwargs) 

99 

100 # Ensure any fields we are *not* using are removed 

101 for field_name in self.fields_to_remove: 

102 self.fields.pop(field_name, None) 

103 

104 def is_exporting(self) -> bool: 

105 """Determine if we are exporting data.""" 

106 return getattr(self, '_exporting_data', False) 

107 

108 def is_field_included( 

109 self, field_name: str, field: OptionalField, kwargs: dict 

110 ) -> bool: 

111 """Determine at runtime whether an OptionalField should be included. 

112 

113 Arguments: 

114 field_name: Name of the field 

115 field: The OptionalField instance 

116 kwargs: The kwargs provided to the serializer instance 

117 

118 Returns: 

119 True if the field should be included, False otherwise. 

120 

121 Order of operations: 

122 

123 - If we are generating the schema, always include the field 

124 - If this is a write request (POST, PUT, PATCH) and we are not exporting, always include the field 

125 - If this is a top-level serializer, check the request query parameters for the filter name 

126 - Check the kwargs provided to the serializer instance 

127 - Finally, fall back to the default_include value for the field itself 

128 """ 

129 field_ref = field.filter_name or field_name 

130 

131 # If we have already found a value for this filter, use it 

132 # This allows multiple optional fields to share the same filter value 

133 cached_value = self.optional_filters.get(field_ref, None) 

134 

135 if cached_value is not None: 

136 return cached_value 

137 

138 # First, check kwargs provided to the serializer instance 

139 # We also pop the value to avoid issues with nested serializers 

140 value = kwargs.pop(field_ref, None) 

141 

142 # We do not want to pop fields while generating the schema 

143 if InvenTree.ready.isGeneratingSchema(): 143 ↛ 144line 143 didn't jump to line 144 because the condition on line 143 was never true

144 return True 

145 

146 if value is not None: 

147 # Cache the value for future reference 

148 self.optional_filters[field_ref] = value 

149 

150 field_kwargs = field.serializer_kwargs or {} 

151 

152 # Skip filtering for a write request - all fields should be present for data creation 

153 if method := getattr(self.request, 'method', None): 

154 if method not in SAFE_METHODS and not self.is_exporting(): 

155 return True 

156 else: 

157 # Ignore write_only fields for read requests 

158 if field_kwargs.get('write_only', False): 158 ↛ 159line 158 didn't jump to line 159 because the condition on line 158 was never true

159 return False 

160 

161 # For a top-level serializer, check request query parameters 

162 if self.request and self.filter_on_query and field.filter_by_query: 

163 param_value = self.request.query_params.get(field_ref, None) 

164 

165 if param_value is not None: 

166 # Convert from list to single value if needed 

167 if type(param_value) == list and len(param_value) == 1: 167 ↛ 168line 167 didn't jump to line 168 because the condition on line 167 was never true

168 param_value = param_value[0] 

169 

170 value = str2bool(param_value) 

171 

172 # Cache the value for future reference 

173 self.optional_filters[field_ref] = value 

174 

175 if value is None: 

176 value = field.default_include 

177 

178 return value 

179 

180 def find_optional_fields(self): 

181 """Find all optional fields defined on this serializer.""" 

182 optional_fields = {} 

183 

184 # Walk upwards through the class hierarchy 

185 seen_vars = set() 

186 

187 for base in self.__class__.__mro__: 

188 for field_name, field in vars(base).items(): 

189 if field_name in seen_vars: 

190 continue 

191 

192 seen_vars.add(field_name) 

193 

194 if field and isinstance(field, OptionalField): 

195 optional_fields[field_name] = field 

196 

197 return optional_fields 

198 

199 def gather_optional_fields(self, kwargs): 

200 """Determine which optional fields will be included on this serializer. 

201 

202 Note that there may be instances of OptionalField in the field set, 

203 which need to either be instantiated or removed. 

204 """ 

205 self.optional_filters = {} 

206 self.prefetch_list = set() 

207 self.fields_to_remove = set() 

208 self.optional_fields = set() 

209 

210 for field_name, field in self.find_optional_fields().items(): 

211 if self.is_field_included(field_name, field, kwargs): 

212 self.optional_fields.add(field_name) 

213 # Add prefetch information 

214 if field.prefetch_fields: 

215 for pf in field.prefetch_fields: 

216 self.prefetch_list.add(pf) 

217 else: 

218 self.fields_to_remove.add(field_name) 

219 

220 def get_field_names(self, declared_fields, info): 

221 """Remove unused fields before returning field names.""" 

222 field_names = super().get_field_names(declared_fields, info) 

223 

224 # Add any optional fields which are included 

225 for field_name in self.optional_fields: 

226 if field_name not in field_names: 

227 field_names.append(field_name) 

228 

229 # Remove any fields which are marked for removal 

230 for field_name in self.fields_to_remove: 

231 if field_name in field_names: 

232 field_names.remove(field_name) 

233 

234 return field_names 

235 

236 def build_optional_field(self, field_name: str): 

237 """Build an optional field, based on the provided field name.""" 

238 field = getattr(self, field_name, None) 

239 

240 if field and isinstance(field, OptionalField): 240 ↛ exitline 240 didn't return from function 'build_optional_field' because the condition on line 240 was always true

241 serializer_kwargs = {**field.serializer_kwargs} or {} 

242 return field.serializer_class, serializer_kwargs 

243 

244 def build_relational_field(self, field_name, relation_info): 

245 """Handle a special case where an OptionalField shadows a model relation.""" 

246 if field_name in self.optional_fields: 

247 if field := self.build_optional_field(field_name): 247 ↛ 250line 247 didn't jump to line 250 because the condition on line 247 was always true

248 return field 

249 

250 return super().build_relational_field(field_name, relation_info) 

251 

252 def build_property_field(self, field_name, model_class): 

253 """Handle a special case where an OptionalField shadows a model property.""" 

254 if field_name in self.optional_fields: 

255 if field := self.build_optional_field(field_name): 255 ↛ 258line 255 didn't jump to line 258 because the condition on line 255 was always true

256 return field 

257 

258 return super().build_property_field(field_name, model_class) 

259 

260 def build_unknown_field(self, field_name, model_class): 

261 """Perform lazy initialization of OptionalFields. 

262 

263 The DRF framework calls this method when it encounters a field which is not yet initialized. 

264 """ 

265 if field := self.build_optional_field(field_name): 265 ↛ 268line 265 didn't jump to line 268 because the condition on line 265 was always true

266 return field 

267 

268 return super().build_unknown_field(field_name, model_class) 

269 

270 def prefetch_queryset(self, queryset: QuerySet) -> QuerySet: 

271 """Apply any prefetching to the queryset based on the optionally included fields. 

272 

273 Args: 

274 queryset: The original queryset. 

275 

276 Returns: 

277 The modified queryset with prefetching applied. 

278 """ 

279 # If we are inside an OPTIONS request, DO NOT PREFETCH 

280 if request := getattr(self, 'request', None): 280 ↛ 288line 280 didn't jump to line 288 because the condition on line 280 was always true

281 if method := getattr(request, 'method', None): 281 ↛ 285line 281 didn't jump to line 285 because the condition on line 281 was always true

282 if str(method).lower() == 'options': 

283 return queryset 

284 

285 if getattr(request, '_metadata_requested', False): 

286 return queryset 

287 

288 if self.prefetch_list and len(self.prefetch_list) > 0: 

289 queryset = queryset.prefetch_related(*list(self.prefetch_list)) 

290 

291 return queryset 

292 

293 

294class EmptySerializer(serializers.Serializer): 

295 """Empty serializer for use in testing.""" 

296 

297 

298class TreePathSerializer(serializers.Serializer): 

299 """Serializer field for representing a tree path.""" 

300 

301 class Meta: 

302 """Metaclass options.""" 

303 

304 fields = [ 

305 'pk', 

306 'name', 

307 # Any fields after this point are optional, and can be included via extra_fields 

308 'icon', 

309 ] 

310 

311 def __init__(self, *args, extra_fields: Optional[list[str]] = None, **kwargs): 

312 """Initialize the TreePathSerializer.""" 

313 super().__init__(*args, **kwargs) 

314 

315 allowed_fields = ['pk', 'name', *(extra_fields or [])] 

316 

317 if InvenTree.ready.isGeneratingSchema(): 317 ↛ 318line 317 didn't jump to line 318 because the condition on line 317 was never true

318 return 

319 

320 for field in list(self.fields.keys()): 

321 if field not in allowed_fields: 

322 self.fields.pop(field, None) 

323 

324 pk = serializers.IntegerField(read_only=True) 

325 name = serializers.CharField(read_only=True) 

326 icon = serializers.CharField(required=False, read_only=True, allow_null=True) 

327 

328 

329class InvenTreeMoneySerializer(MoneyField): 

330 """Custom serializer for 'MoneyField', which ensures that passed values are numerically valid. 

331 

332 Ref: https://github.com/django-money/django-money/blob/master/djmoney/contrib/django_rest_framework/fields.py 

333 """ 

334 

335 def __init__(self, *args, **kwargs): 

336 """Override default values.""" 

337 kwargs['max_digits'] = kwargs.get('max_digits', 19) 

338 self.decimal_places = kwargs['decimal_places'] = kwargs.get('decimal_places', 6) 

339 kwargs['required'] = kwargs.get('required', False) 

340 

341 super().__init__(*args, **kwargs) 

342 

343 def to_representation(self, obj): 

344 """Convert the Money object to a decimal value for representation.""" 

345 val = super().to_representation(obj) 

346 

347 return float(val) 

348 

349 def get_value(self, data): 

350 """Test that the returned amount is a valid Decimal.""" 

351 amount = super(DecimalField, self).get_value(data) 

352 

353 # Convert an empty string to None 

354 if len(str(amount).strip()) == 0: 

355 amount = None 

356 

357 try: 

358 if amount is not None and amount is not empty: 

359 # Convert to a Decimal instance, and round to maximum allowed decimal places 

360 amount = Decimal(amount) 

361 amount = round(amount, self.decimal_places) 

362 except Exception: 

363 raise ValidationError({self.field_name: [_('Must be a valid number')]}) 

364 

365 currency = data.get( 

366 get_currency_field_name(self.field_name), self.default_currency 

367 ) 

368 

369 if ( 369 ↛ 375line 369 didn't jump to line 375 because the condition on line 369 was never true

370 currency 

371 and amount is not None 

372 and not isinstance(amount, MONEY_CLASSES) 

373 and amount is not empty 

374 ): 

375 return Money(amount, currency) 

376 

377 try: 

378 fp_amount = float(amount) 

379 return fp_amount 

380 except Exception: 

381 return amount 

382 

383 

384@extend_schema_field(serializers.CharField()) 

385class InvenTreeCurrencySerializer(serializers.ChoiceField): 

386 """Custom serializers for selecting currency option.""" 

387 

388 def __init__(self, *args, **kwargs): 

389 """Initialize the currency serializer.""" 

390 choices = currency_code_mappings() 

391 

392 allow_blank = kwargs.get('allow_blank', False) or kwargs.get( 

393 'allow_null', False 

394 ) 

395 

396 if allow_blank: 

397 choices = [('', '---------'), *choices] 

398 

399 kwargs['choices'] = choices 

400 

401 if 'default' not in kwargs and 'required' not in kwargs: 

402 kwargs['default'] = '' if allow_blank else currency_code_default 

403 

404 if 'label' not in kwargs: 

405 kwargs['label'] = _('Currency') 

406 

407 if 'help_text' not in kwargs: 

408 kwargs['help_text'] = _('Select currency from available options') 

409 

410 if InvenTree.ready.isGeneratingSchema(): 410 ↛ 411line 410 didn't jump to line 411 because the condition on line 410 was never true

411 kwargs['help_text'] = ( 

412 kwargs['help_text'] 

413 + '\n\n' 

414 + '\n'.join(f'* `{value}` - {label}' for value, label in choices) 

415 + "\n\nOther valid currencies may be found in the 'CURRENCY_CODES' global setting." 

416 ) 

417 

418 super().__init__(*args, **kwargs) 

419 

420 

421class DependentField(serializers.Field): 

422 """A dependent field can be used to dynamically return child fields based on the value of other fields.""" 

423 

424 child = None 

425 

426 def __init__(self, *args, depends_on, field_serializer, **kwargs): 

427 """A dependent field can be used to dynamically return child fields based on the value of other fields. 

428 

429 Example: 

430 This example adds two fields. If the client selects integer, an integer field will be shown, but if he 

431 selects char, an char field will be shown. For any other value, nothing will be shown. 

432 

433 class TestSerializer(serializers.Serializer): 

434 select_type = serializers.ChoiceField(choices=[ 

435 ("integer", "Integer"), 

436 ("char", "Char"), 

437 ]) 

438 my_field = DependentField(depends_on=["select_type"], field_serializer="get_my_field") 

439 

440 def get_my_field(self, fields): 

441 if fields["select_type"] == "integer": 

442 return serializers.IntegerField() 

443 if fields["select_type"] == "char": 

444 return serializers.CharField() 

445 """ 

446 super().__init__(*args, **kwargs) 

447 

448 self.depends_on = depends_on 

449 self.field_serializer = field_serializer 

450 

451 def get_child(self, raise_exception=False): 

452 """This method tries to extract the child based on the provided data in the request by the client.""" 

453 data = deepcopy(self.context['request'].data) 

454 

455 def visit_parent(node): 

456 """Recursively extract the data for the parent field/serializer in reverse.""" 

457 nonlocal data 

458 

459 if node.parent: 

460 visit_parent(node.parent) 

461 

462 # only do for composite fields and stop right before the current field 

463 if hasattr(node, 'child') and node is not self and isinstance(data, dict): 

464 data = data.get(node.field_name, None) 

465 

466 visit_parent(self) 

467 

468 # ensure that data is a dictionary and that a parent exists 

469 if not isinstance(data, dict) or self.parent is None: 

470 return 

471 

472 # check if the request data contains the dependent fields, otherwise skip getting the child 

473 for f in self.depends_on: 

474 if data.get(f, None) is None: 

475 if ( 

476 self.parent 

477 and (v := getattr(self.parent.fields[f], 'default', None)) 

478 is not None 

479 ): 

480 data[f] = v 

481 else: 

482 return 

483 

484 # partially validate the data for options requests that set raise_exception while calling .get_child(...) 

485 if raise_exception: 

486 validation_data = {k: v for k, v in data.items() if k in self.depends_on} 

487 serializer = self.parent.__class__( 

488 context=self.context, data=validation_data, partial=True 

489 ) 

490 serializer.is_valid(raise_exception=raise_exception) 

491 

492 # try to get the field serializer 

493 field_serializer = getattr(self.parent, self.field_serializer) 

494 child = field_serializer(data) 

495 

496 if not child: 

497 return 

498 

499 self.child = child 

500 self.child.bind(field_name='', parent=self) 

501 

502 def to_internal_value(self, data): 

503 """This method tries to convert the data to an internal representation based on the defined to_internal_value method on the child.""" 

504 self.get_child() 

505 if self.child: 

506 return self.child.to_internal_value(data) 

507 

508 return None 

509 

510 def to_representation(self, value): 

511 """This method tries to convert the data to representation based on the defined to_representation method on the child.""" 

512 self.get_child() 

513 if self.child: 

514 return self.child.to_representation(value) 

515 

516 return None 

517 

518 

519class InvenTreeModelSerializer(serializers.ModelSerializer): 

520 """Inherits the standard Django ModelSerializer class, but also ensures that the underlying model class data are checked on validation.""" 

521 

522 # Switch out URLField mapping 

523 serializer_field_mapping = { 

524 **serializers.ModelSerializer.serializer_field_mapping, 

525 models.URLField: InvenTreeRestURLField, 

526 InvenTreeURLField: InvenTreeRestURLField, 

527 } 

528 

529 def __init__(self, instance=None, data=empty, **kwargs): 

530 """Custom __init__ routine to ensure that *default* values (as specified in the ORM) are used by the DRF serializers, *if* the values are not provided by the user.""" 

531 # If instance is None, we are creating a new instance 

532 if instance is None and data is not empty: 

533 if data is None: 

534 data = OrderedDict() 

535 else: 

536 new_data = OrderedDict() 

537 new_data.update(data) 

538 

539 data = new_data 

540 

541 # Add missing fields which have default values 

542 ModelClass = self.Meta.model 

543 

544 fields = model_meta.get_field_info(ModelClass) 

545 

546 for field_name, field in fields.fields.items(): 

547 """ 

548 Update the field IF (and ONLY IF): 

549 

550 - The field has a specified default value 

551 - The field does not already have a value set 

552 """ 

553 if field.has_default() and field_name not in data: 

554 value = field.default 

555 

556 # Account for callable functions 

557 if callable(value): 

558 try: 

559 value = value() 

560 except Exception: 

561 continue 

562 

563 data[field_name] = value 

564 

565 super().__init__(instance, data, **kwargs) 

566 

567 def get_initial(self): 

568 """Construct initial data for the serializer. 

569 

570 Use the 'default' values specified by the django model definition 

571 """ 

572 initials = super().get_initial().copy() 

573 

574 # Are we creating a new instance? 

575 if self.instance is None: 

576 ModelClass = self.Meta.model 

577 

578 fields = model_meta.get_field_info(ModelClass) 

579 

580 for field_name, field in fields.fields.items(): 

581 if field.has_default() and field_name not in initials: 

582 value = field.default 

583 

584 # Account for callable functions 

585 if callable(value): 

586 try: 

587 value = value() 

588 except Exception: 

589 continue 

590 

591 initials[field_name] = value 

592 

593 return initials 

594 

595 def skip_create_fields(self): 

596 """Return a list of 'fields' which should be skipped for model creation. 

597 

598 This is used to 'bypass' a shortcoming of the DRF framework, 

599 which does not allow us to have writeable serializer fields which do not exist on the model. 

600 

601 Default implementation returns an empty list 

602 """ 

603 return [] 

604 

605 def save(self, **kwargs): 

606 """Catch any django ValidationError thrown at the moment `save` is called, and re-throw as a DRF ValidationError.""" 

607 try: 

608 super().save(**kwargs) 

609 except (ValidationError, DjangoValidationError) as exc: 

610 raise ValidationError(detail=serializers.as_serializer_error(exc)) 

611 

612 return self.instance 

613 

614 def create(self, validated_data): 

615 """Custom create method which supports field adjustment.""" 

616 initial_data = validated_data.copy() 

617 

618 # Remove any fields which do not exist on the model 

619 for field in self.skip_create_fields(): 

620 initial_data.pop(field, None) 

621 

622 return super().create(initial_data) 

623 

624 def update(self, instance, validated_data): 

625 """Catch any django ValidationError, and re-throw as a DRF ValidationError.""" 

626 try: 

627 instance = super().update(instance, validated_data) 

628 except (ValidationError, DjangoValidationError) as exc: 

629 raise ValidationError(detail=serializers.as_serializer_error(exc)) 

630 

631 return instance 

632 

633 def run_validation(self, data=empty): 

634 """Perform serializer validation. 

635 

636 In addition to running validators on the serializer fields, 

637 this class ensures that the underlying model is also validated. 

638 """ 

639 # Run any native validation checks first (may raise a ValidationError) 

640 data = super().run_validation(data) 

641 

642 if not hasattr(self, 'instance') or self.instance is None: 

643 # No instance exists (we are creating a new one) 

644 

645 initial_data = data.copy() 

646 

647 for field in self.skip_create_fields(): 

648 # Remove any fields we do not wish to provide to the model 

649 initial_data.pop(field, None) 

650 

651 # Create a (RAM only) instance for extra testing 

652 instance = self.Meta.model(**initial_data) 

653 else: 

654 # Instance already exists (we are updating!) 

655 instance = self.instance 

656 

657 # Update instance fields 

658 for attr, value in data.items(): 

659 try: 

660 setattr(instance, attr, value) 

661 except (ValidationError, DjangoValidationError) as exc: 

662 raise ValidationError(detail=serializers.as_serializer_error(exc)) 

663 

664 # Run a 'full_clean' on the model. 

665 # Note that by default, DRF does *not* perform full model validation! 

666 try: 

667 instance.full_clean() 

668 except (ValidationError, DjangoValidationError) as exc: 

669 if hasattr(exc, 'message_dict'): 669 ↛ 671line 669 didn't jump to line 671 because the condition on line 669 was always true

670 data = {**exc.message_dict} 

671 elif hasattr(exc, 'message'): 

672 data = {'non_field_errors': [str(exc.message)]} 

673 else: 

674 data = {'non_field_errors': [str(exc)]} 

675 

676 # Change '__all__' key (django style) to 'non_field_errors' (DRF style) 

677 if hasattr(data, '__all__'): 677 ↛ 678line 677 didn't jump to line 678 because the condition on line 677 was never true

678 data['non_field_errors'] = data.pop('__all__') 

679 

680 raise ValidationError(data) 

681 

682 return data 

683 

684 

685class InvenTreeTaggitSerializer(TaggitSerializer): 

686 """Updated from https://github.com/glemmaPaul/django-taggit-serializer.""" 

687 

688 def update(self, instance, validated_data): 

689 """Overridden update method to re-add the tagmanager.""" 

690 to_be_tagged, validated_data = self._pop_tags(validated_data) 

691 

692 tag_object = super().update(instance, validated_data) 

693 

694 for key in to_be_tagged: 694 ↛ 696line 694 didn't jump to line 696 because the loop on line 694 never started

695 # re-add the tagmanager 

696 new_tagobject = tag_object.__class__.objects.get(id=tag_object.id) 

697 setattr(tag_object, key, getattr(new_tagobject, key)) 

698 

699 return self._save_tags(tag_object, to_be_tagged) 

700 

701 

702class InvenTreeAttachmentSerializerField(serializers.FileField): 

703 """Override the DRF native FileField serializer, to remove the leading server path. 

704 

705 For example, the FileField might supply something like: 

706 

707 http://127.0.0.1:8000/media/foo/bar.jpg 

708 

709 Whereas we wish to return: 

710 

711 /media/foo/bar.jpg 

712 

713 If the server process is serving the data at 127.0.0.1, 

714 but a proxy service (e.g. nginx) is then providing DNS lookup to the outside world, 

715 then an attachment which prefixes the "address" of the internal server 

716 will not be accessible from the outside world. 

717 """ 

718 

719 def to_representation(self, value): 

720 """To json-serializable type.""" 

721 if not value: 

722 return None 

723 

724 return default_storage.url(str(value)) 

725 

726 

727class InvenTreeImageSerializerField(serializers.ImageField): 

728 """Custom image serializer. 

729 

730 On upload, validate that the file is a valid image file 

731 """ 

732 

733 def to_representation(self, value): 

734 """To json-serializable type.""" 

735 if not value: 735 ↛ 738line 735 didn't jump to line 738 because the condition on line 735 was always true

736 return None 

737 

738 return default_storage.url(str(value)) 

739 

740 

741class InvenTreeDecimalField(serializers.FloatField): 

742 """Custom serializer for decimal fields. 

743 

744 Solves the following issues: 

745 - The normal DRF DecimalField renders values with trailing zeros 

746 - Using a FloatField can result in rounding issues: https://code.djangoproject.com/ticket/30290 

747 """ 

748 

749 def to_internal_value(self, data): 

750 """Convert to python type.""" 

751 if data in [None, '']: 751 ↛ 752line 751 didn't jump to line 752 because the condition on line 751 was never true

752 if self.allow_null: 

753 return None 

754 raise serializers.ValidationError(_('This field may not be null.')) 

755 

756 # Convert the value to a string, and then a decimal 

757 try: 

758 return Decimal(str(data)) 

759 except Exception: 

760 raise serializers.ValidationError(_('Invalid value')) 

761 

762 

763class CustomStatusSerializerMixin(serializers.Serializer): 

764 """Serializer mixin for models that support custom status values. 

765 

766 Provides a `status_text` SerializerMethodField that resolves custom 

767 status labels with a single database query per model per serializer 

768 context (i.e. one query for a whole list page) rather than one query per 

769 object (N+1). 

770 """ 

771 

772 status_text = serializers.SerializerMethodField() 

773 

774 @extend_schema_field(serializers.CharField(allow_null=True)) 

775 def get_status_text(self, instance) -> Optional[str]: 

776 """Return the human-readable status text for the instance. 

777 

778 Uses a per-context cache keyed by model name so that all objects in a 

779 single serialization pass share one DB hit for custom label lookup. 

780 

781 During write operations DRF may call to_representation on the raw 

782 validated_data dict rather than a model instance (e.g. when building 

783 response headers). Return None in that case — the response body is 

784 always produced from a real instance via a separate serializer call. 

785 """ 

786 if not hasattr(instance, 'get_custom_status'): 786 ↛ 787line 786 didn't jump to line 787 because the condition on line 786 was never true

787 return None 

788 

789 custom_key = instance.get_custom_status() 

790 

791 if custom_key is None: 791 ↛ 794line 791 didn't jump to line 794 because the condition on line 791 was always true

792 return instance.status_class.label(instance.get_status()) 

793 

794 model_name = instance._meta.model_name 

795 cache_key = f'_custom_status_labels_{model_name}' 

796 

797 # Cache a dict of custom status labels for this model, if not already cached 

798 if cache_key not in self.context: 

799 from common.models import InvenTreeCustomUserStateModel 

800 

801 self.context[cache_key] = { 

802 obj.key: obj.label 

803 for obj in InvenTreeCustomUserStateModel.objects.filter( 

804 model__model=model_name 

805 ) 

806 } 

807 

808 return self.context[cache_key].get( 

809 custom_key, instance.status_class.label(instance.get_status()) 

810 ) 

811 

812 

813class NotesFieldMixin: 

814 """Serializer mixin for handling 'notes' fields. 

815 

816 The 'notes' field will be hidden in a LIST serializer, 

817 but available in a DETAIL serializer. 

818 """ 

819 

820 def __init__(self, *args, **kwargs): 

821 """Remove 'notes' field from list views.""" 

822 super().__init__(*args, **kwargs) 

823 

824 if hasattr(self, 'context'): 824 ↛ exitline 824 didn't return from function '__init__' because the condition on line 824 was always true

825 request = self.context.get('request', None) 

826 method = getattr(request, 'method', None) 

827 

828 if view := self.context.get('view', None): 

829 if ( 

830 issubclass(view.__class__, ListModelMixin) 

831 and method in SAFE_METHODS 

832 and not InvenTree.ready.isGeneratingSchema() 

833 ): 

834 self.fields.pop('notes', None) 

835 

836 

837class ContentTypeField(serializers.ChoiceField): 

838 """Serializer field which represents a ContentType as 'app_label.model_name'. 

839 

840 This field converts a ContentType instance to a string representation in the format 'app_label.model_name' during serialization, and vice versa during deserialization. 

841 

842 Additionally, a "mixin_class" can be supplied to the field, which will restrict the valid content types to only those models which inherit from the specified mixin. 

843 """ 

844 

845 mixin_class = None 

846 

847 def __init__(self, *args, mixin_class=None, **kwargs): 

848 """Initialize the ContentTypeField. 

849 

850 Args: 

851 mixin_class: Optional mixin class to restrict valid content types. 

852 """ 

853 from InvenTree.cache import get_cached_content_types 

854 

855 self.mixin_class = mixin_class 

856 

857 # Override the 'choices' field, to limit to the appropriate models 

858 if self.mixin_class is not None: 858 ↛ 869line 858 didn't jump to line 869 because the condition on line 858 was always true

859 models = getModelsWithMixin(self.mixin_class) 

860 

861 kwargs['choices'] = [ 

862 ( 

863 f'{model._meta.app_label}.{model._meta.model_name}', 

864 model._meta.verbose_name, 

865 ) 

866 for model in models 

867 ] 

868 else: 

869 content_types = get_cached_content_types() 

870 

871 kwargs['choices'] = [ 

872 (f'{ct.app_label}.{ct.model}', str(ct)) for ct in content_types 

873 ] 

874 

875 if kwargs.get('allow_null') or kwargs.get('allow_blank'): 

876 kwargs['choices'] = [('', '---------'), *kwargs['choices']] 

877 

878 super().__init__(*args, **kwargs) 

879 

880 def to_representation(self, value): 

881 """Convert ContentType instance to string representation.""" 

882 return f'{value.app_label}.{value.model}' 

883 

884 def to_internal_value(self, data): 

885 """Convert string representation back to ContentType instance.""" 

886 content_type = None 

887