Coverage for src/backend/InvenTree/common/api.py: 75%

639 statements  

« prev     ^ index     » next       coverage.py v7.15.2, created at 2026-10-07 17:47 +0000

1"""Provides a JSON API for common components.""" 

2 

3import json 

4import json.decoder 

5 

6from django.conf import settings 

7from django.contrib.contenttypes.models import ContentType 

8from django.core.exceptions import FieldDoesNotExist, ValidationError 

9from django.db.models import Q 

10from django.http import JsonResponse 

11from django.http.response import HttpResponse 

12from django.urls import include, path, re_path 

13from django.utils.decorators import method_decorator 

14from django.utils.translation import gettext_lazy as _ 

15from django.views.decorators.cache import cache_control 

16from django.views.decorators.csrf import csrf_exempt 

17 

18import django_filters.rest_framework.filters as rest_filters 

19import django_q.models 

20import django_q.tasks 

21from django_filters.rest_framework.filterset import FilterSet 

22from djmoney.contrib.exchange.models import ExchangeBackend, Rate 

23from drf_spectacular.utils import ( 

24 OpenApiParameter, 

25 OpenApiResponse, 

26 extend_schema, 

27 extend_schema_view, 

28) 

29from error_report.models import Error 

30from opentelemetry import trace 

31from pint._typing import UnitLike 

32from rest_framework import serializers, viewsets 

33from rest_framework.decorators import action 

34from rest_framework.exceptions import NotAcceptable, NotFound, PermissionDenied 

35from rest_framework.permissions import IsAdminUser, IsAuthenticated 

36from rest_framework.response import Response 

37from rest_framework.views import APIView 

38from sql_util.utils import SubqueryCount 

39from taggit.models import Tag 

40 

41import common.filters 

42import common.models 

43import common.serializers 

44import InvenTree.conversion 

45import InvenTree.models 

46import InvenTree.ready 

47from common.icons import get_icon_packs 

48from common.settings import get_global_setting 

49from data_exporter.mixins import DataExportViewMixin 

50from generic.states.api import urlpattern as generic_states_api_urls 

51from InvenTree.api import ( 

52 BulkCreateMixin, 

53 BulkDeleteMixin, 

54 BulkDeleteViewsetMixin, 

55 GenericMetadataView, 

56 SimpleGenericMetadataView, 

57 meta_path, 

58) 

59from InvenTree.config import CONFIG_LOOKUPS 

60from InvenTree.filters import ORDER_FILTER, SEARCH_ORDER_FILTER 

61from InvenTree.helpers import inheritors, str2bool 

62from InvenTree.helpers_api import ( 

63 InvenTreeApiRouter, 

64 RetrieveDestroyModelViewSet, 

65 RetrieveUpdateDestroyModelViewSet, 

66) 

67from InvenTree.helpers_email import send_email 

68from InvenTree.mixins import ( 

69 CreateAPI, 

70 ListAPI, 

71 ListCreateAPI, 

72 OutputOptionsMixin, 

73 RetrieveAPI, 

74 RetrieveUpdateAPI, 

75 RetrieveUpdateDestroyAPI, 

76) 

77from InvenTree.permissions import ( 

78 AllowAnyOrReadScope, 

79 GlobalSettingsPermissions, 

80 IsAdminOrAdminScope, 

81 IsAuthenticatedOrReadScope, 

82 IsStaffOrReadOnlyScope, 

83 IsSuperuserOrSuperScope, 

84 UserSettingsPermissionsOrScope, 

85) 

86from InvenTree.serializers import EmptySerializer 

87 

88admin_router = InvenTreeApiRouter() 

89common_router = InvenTreeApiRouter() 

90 

91 

92class CsrfExemptMixin: 

93 """Exempts the view from CSRF requirements.""" 

94 

95 @method_decorator(csrf_exempt) 

96 def dispatch(self, *args, **kwargs): 

97 """Overwrites dispatch to be exempt from CSRF checks.""" 

98 return super().dispatch(*args, **kwargs) 

99 

100 

101class WebhookView(CsrfExemptMixin, APIView): 

102 """Endpoint for receiving webhooks.""" 

103 

104 authentication_classes = [] 

105 permission_classes = [] 

106 model_class = common.models.WebhookEndpoint 

107 run_async = False 

108 serializer_class = None 

109 

110 @extend_schema( 

111 responses={ 

112 200: OpenApiResponse( 

113 description='Any data can be posted to the endpoint - everything will be passed to the WebhookEndpoint model.' 

114 ) 

115 } 

116 ) 

117 def post(self, request, endpoint, *args, **kwargs): 

118 """Process incoming webhook.""" 

119 # get webhook definition 

120 self._get_webhook(endpoint, request, *args, **kwargs) 

121 

122 # check headers 

123 headers = request.headers 

124 try: 

125 payload = json.loads(request.body) 

126 except json.decoder.JSONDecodeError as error: 

127 raise NotAcceptable(error.msg) 

128 

129 # validate 

130 self.webhook.validate_token(payload, headers, request) 

131 # process data 

132 message = self.webhook.save_data(payload, headers, request) 

133 if self.run_async: 

134 django_q.tasks.async_task(self._process_payload, message.id) 

135 else: 

136 self._process_result( 

137 self.webhook.process_payload(message, payload, headers), message 

138 ) 

139 

140 data = self.webhook.get_return(payload, headers, request) 

141 return HttpResponse(data) 

142 

143 def _process_payload(self, message_id): 

144 message = common.models.WebhookMessage.objects.get(message_id=message_id) 

145 self._process_result( 

146 self.webhook.process_payload(message, message.body, message.header), message 

147 ) 

148 

149 def _process_result(self, result, message): 

150 if result: 

151 message.worked_on = result 

152 message.save() 

153 else: 

154 message.delete() 

155 

156 def _escalate_object(self, obj): 

157 classes = inheritors(obj.__class__) 

158 for cls in classes: 

159 mdl_name = cls._meta.model_name 

160 if hasattr(obj, mdl_name): 

161 return getattr(obj, mdl_name) 

162 return obj 

163 

164 def _get_webhook(self, endpoint, request, *args, **kwargs): 

165 try: 

166 webhook = self.model_class.objects.get(endpoint_id=endpoint) 

167 self.webhook = self._escalate_object(webhook) 

168 self.webhook.init(request, *args, **kwargs) 

169 return self.webhook.process_webhook() 

170 except self.model_class.DoesNotExist: 

171 raise NotFound() 

172 

173 

174class CurrencyViewSet(viewsets.GenericViewSet): 

175 """Viewset for currency exchange information.""" 

176 

177 permission_classes = [IsAuthenticatedOrReadScope] 

178 serializer_class = EmptySerializer 

179 

180 @action( 

181 detail=False, 

182 methods=['get'], 

183 serializer_class=common.serializers.CurrencyExchangeSerializer, 

184 ) 

185 def exchange(self, request, fmt=None): 

186 """Return information on available currency conversions.""" 

187 # Extract a list of all available rates 

188 try: 

189 rates = Rate.objects.all() 

190 except Exception: 

191 rates = [] 

192 

193 # Information on last update 

194 try: 

195 backend = ExchangeBackend.objects.filter(name='InvenTreeExchange') 

196 

197 if backend.exists(): 197 ↛ 201line 197 didn't jump to line 201 because the condition on line 197 was always true

198 backend = backend.first() 

199 updated = backend.last_update 

200 else: 

201 updated = None 

202 except Exception: 

203 updated = None 

204 

205 response = { 

206 'base_currency': get_global_setting( 

207 'INVENTREE_DEFAULT_CURRENCY', backup_value='USD' 

208 ), 

209 'exchange_rates': {}, 

210 'updated': updated, 

211 } 

212 

213 for rate in rates: 

214 response['exchange_rates'][rate.currency] = rate.value 

215 

216 return Response(response) 

217 

218 @action( 

219 detail=False, 

220 methods=['post'], 

221 permission_classes=[IsAuthenticatedOrReadScope, IsAdminUser], 

222 ) 

223 def refresh(self, request, *args, **kwargs): 

224 """Performing a POST request will update currency exchange rates.""" 

225 from InvenTree.tasks import update_exchange_rates 

226 

227 update_exchange_rates(force=True) 

228 

229 return Response({'success': 'Exchange rates updated'}) 

230 

231 

232common_router.register('currency', CurrencyViewSet, basename='api-currency') 

233 

234 

235class SettingsList(ListAPI): 

236 """Generic ListView for settings. 

237 

238 This is inherited by all list views for settings. 

239 """ 

240 

241 filter_backends = SEARCH_ORDER_FILTER 

242 

243 ordering_fields = ['pk', 'key', 'name'] 

244 

245 search_fields = ['key'] 

246 

247 

248class GlobalSettingsList(SettingsList): 

249 """API endpoint for accessing a list of global settings objects.""" 

250 

251 queryset = common.models.InvenTreeSetting.objects.exclude(key__startswith='_') 

252 serializer_class = common.serializers.GlobalSettingsSerializer 

253 permission_classes = [IsAuthenticated, GlobalSettingsPermissions] 

254 

255 def list(self, request, *args, **kwargs): 

256 """Ensure all global settings are created.""" 

257 common.models.InvenTreeSetting.build_default_values() 

258 return super().list(request, *args, **kwargs) 

259 

260 

261class GlobalSettingsDetail(RetrieveUpdateAPI): 

262 """Detail view for an individual "global setting" object. 

263 

264 - User must have 'staff' status to view / edit 

265 """ 

266 

267 lookup_field = 'key' 

268 queryset = common.models.InvenTreeSetting.objects.exclude(key__startswith='_') 

269 serializer_class = common.serializers.GlobalSettingsSerializer 

270 permission_classes = [IsAuthenticated, GlobalSettingsPermissions] 

271 

272 def get_object(self): 

273 """Attempt to find a global setting object with the provided key.""" 

274 key = str(self.kwargs['key']).upper() 

275 

276 if key.startswith('_') or key not in common.models.InvenTreeSetting.SETTINGS: 

277 raise NotFound() 

278 

279 return common.models.InvenTreeSetting.get_setting_object( 

280 key, cache=False, create=True 

281 ) 

282 

283 

284class UserSettingsList(SettingsList): 

285 """API endpoint for accessing a list of user settings objects.""" 

286 

287 queryset = common.models.InvenTreeUserSetting.objects.all() 

288 serializer_class = common.serializers.UserSettingsSerializer 

289 permission_classes = [UserSettingsPermissionsOrScope] 

290 

291 def list(self, request, *args, **kwargs): 

292 """Ensure all user settings are created.""" 

293 common.models.InvenTreeUserSetting.build_default_values(user=request.user) 

294 return super().list(request, *args, **kwargs) 

295 

296 def get_queryset(self): 

297 """Return prefetched queryset.""" 

298 queryset = super().get_queryset() 

299 queryset = queryset.prefetch_related('user') 

300 

301 return queryset 

302 

303 def filter_queryset(self, queryset): 

304 """Only list settings which apply to the current user.""" 

305 try: 

306 user = self.request.user 

307 except AttributeError: # pragma: no cover 

308 return common.models.InvenTreeUserSetting.objects.none() 

309 

310 queryset = super().filter_queryset(queryset) 

311 

312 if not user.is_authenticated: # pragma: no cover 312 ↛ 313line 312 didn't jump to line 313 because the condition on line 312 was never true

313 raise PermissionDenied('User must be authenticated to access user settings') 

314 

315 queryset = queryset.filter(user=user) 

316 

317 return queryset 

318 

319 

320class UserSettingsDetail(RetrieveUpdateAPI): 

321 """Detail view for an individual "user setting" object. 

322 

323 - User can only view / edit settings their own settings objects 

324 """ 

325 

326 lookup_field = 'key' 

327 queryset = common.models.InvenTreeUserSetting.objects.all() 

328 serializer_class = common.serializers.UserSettingsSerializer 

329 permission_classes = [UserSettingsPermissionsOrScope] 

330 

331 def get_object(self): 

332 """Attempt to find a user setting object with the provided key.""" 

333 key = str(self.kwargs['key']).upper() 

334 

335 if ( 

336 key.startswith('_') 

337 or key not in common.models.InvenTreeUserSetting.SETTINGS 

338 ): 

339 raise NotFound() 

340 

341 return common.models.InvenTreeUserSetting.get_setting_object( 

342 key, user=self.request.user, cache=False, create=True 

343 ) 

344 

345 

346class NotificationMessageViewSet( 

347 BulkDeleteViewsetMixin, RetrieveUpdateDestroyModelViewSet 

348): 

349 """Notifications for the current user. 

350 

351 - User can only view / delete their own notification objects 

352 """ 

353 

354 queryset = common.models.NotificationMessage.objects.all() 

355 serializer_class = common.serializers.NotificationMessageSerializer 

356 permission_classes = [UserSettingsPermissionsOrScope] 

357 

358 filter_backends = SEARCH_ORDER_FILTER 

359 ordering_fields = ['category', 'name', 'read', 'creation'] 

360 search_fields = ['name', 'message'] 

361 filterset_fields = ['category', 'read'] 

362 

363 def get_queryset(self): 

364 """Return prefetched queryset.""" 

365 queryset = ( 

366 super() 

367 .get_queryset() 

368 .prefetch_related( 

369 'source_content_type', 

370 'source_object', 

371 'target_content_type', 

372 'target_object', 

373 'user', 

374 ) 

375 ) 

376 

377 return queryset 

378 

379 def filter_queryset(self, queryset): 

380 """Only list notifications which apply to the current user.""" 

381 try: 

382 user = self.request.user 

383 except AttributeError: 

384 return common.models.NotificationMessage.objects.none() 

385 

386 queryset = super().filter_queryset(queryset) 

387 

388 if not user.is_authenticated: # pragma: no cover 388 ↛ 389line 388 didn't jump to line 389 because the condition on line 388 was never true

389 raise PermissionDenied('User must be authenticated to access notifications') 

390 

391 queryset = queryset.filter(user=user) 

392 return queryset 

393 

394 def filter_delete_queryset(self, queryset, request): 

395 """Ensure that the user can only delete their *own* notifications.""" 

396 queryset = queryset.filter(user=request.user) 

397 return queryset 

398 

399 @action( 

400 detail=False, methods=['post'], permission_classes=[IsAuthenticatedOrReadScope] 

401 ) 

402 def readall(self, request, *args, **kwargs): 

403 """Set all messages for the current user as read.""" 

404 try: 

405 self.queryset.filter(user=request.user, read=False).update(read=True) 

406 return Response({'status': 'ok'}) 

407 except Exception as exc: 

408 raise serializers.ValidationError( 

409 detail=serializers.as_serializer_error(exc) 

410 ) 

411 

412 

413common_router.register( 

414 'notifications', NotificationMessageViewSet, basename='api-notifications' 

415) 

416 

417 

418class NewsFeedViewSet(BulkDeleteViewsetMixin, RetrieveUpdateDestroyModelViewSet): 

419 """Newsfeed from the official inventree.org website.""" 

420 

421 queryset = common.models.NewsFeedEntry.objects.all() 

422 serializer_class = common.serializers.NewsFeedEntrySerializer 

423 permission_classes = [IsAdminOrAdminScope] 

424 

425 filter_backends = ORDER_FILTER 

426 ordering = '-published' 

427 ordering_fields = ['published', 'author', 'read'] 

428 filterset_fields = ['read'] 

429 

430 

431common_router.register('news', NewsFeedViewSet, basename='api-news') 

432 

433 

434@extend_schema_view( 

435 retrieve=extend_schema( 

436 parameters=[ 

437 OpenApiParameter( 

438 name='key', 

439 description='Unique identifier for this configuration', 

440 required=True, 

441 location=OpenApiParameter.PATH, 

442 ) 

443 ] 

444 ) 

445) 

446class ConfigViewSet(viewsets.ReadOnlyModelViewSet): 

447 """All accessed/in-use configurations.""" 

448 

449 queryset = CONFIG_LOOKUPS 

450 serializer_class = common.serializers.ConfigSerializer 

451 permission_classes = [IsSuperuserOrSuperScope] 

452 lookup_field = 'key' 

453 

454 # Specifically disable pagination for this view 

455 pagination_class = None 

456 

457 def get_object(self): 

458 """Attempt to find a config object with the provided key.""" 

459 key = self.kwargs['key'] 

460 value = CONFIG_LOOKUPS.get(key, None) 

461 if not value: 

462 raise NotFound() 

463 return {key: value} 

464 

465 

466admin_router.register('config', ConfigViewSet, basename='api-config') 

467 

468 

469class NotesImageList(ListCreateAPI): 

470 """List view for all notes images.""" 

471 

472 queryset = common.models.NotesImage.objects.all() 

473 serializer_class = common.serializers.NotesImageSerializer 

474 permission_classes = [IsAuthenticatedOrReadScope] 

475 

476 filter_backends = SEARCH_ORDER_FILTER 

477 

478 search_fields = ['user', 'model_type', 'model_id'] 

479 

480 def perform_create(self, serializer): 

481 """Create (upload) a new notes image.""" 

482 serializer.save(user=self.request.user) 

483 

484 

485class ProjectCodeList(DataExportViewMixin, ListCreateAPI): 

486 """List view for all project codes.""" 

487 

488 queryset = common.models.ProjectCode.objects.all() 

489 serializer_class = common.serializers.ProjectCodeSerializer 

490 permission_classes = [IsStaffOrReadOnlyScope] 

491 filter_backends = SEARCH_ORDER_FILTER 

492 

493 ordering_fields = ['code'] 

494 

495 search_fields = ['code', 'description'] 

496 

497 

498class ProjectCodeDetail(RetrieveUpdateDestroyAPI): 

499 """Detail view for a particular project code.""" 

500 

501 queryset = common.models.ProjectCode.objects.all() 

502 serializer_class = common.serializers.ProjectCodeSerializer 

503 permission_classes = [IsStaffOrReadOnlyScope] 

504 

505 

506class TagFilter(FilterSet): 

507 """Custom filters for the TagList API endpoint.""" 

508 

509 class Meta: 

510 """Metaclass options for the filterset.""" 

511 

512 model = Tag 

513 fields = [] 

514 

515 model_type = rest_filters.CharFilter(method='filter_model_type', label='Model Type') 

516 

517 def filter_model_type(self, queryset, name, value): 

518 """Filter to tags which have been applied to the given model type.""" 

519 ct = common.filters.determine_content_type(value) 

520 

521 if ct is None: 

522 raise ValidationError({'model_type': f'Invalid model type: {value}'}) 

523 

524 return queryset.filter(taggit_taggeditem_items__content_type=ct).distinct() 

525 

526 

527class TagMixin: 

528 """Mixin class for Tag views.""" 

529 

530 serializer_class = common.serializers.TagSerializer 

531 queryset = Tag.objects.all() 

532 permission_classes = [IsStaffOrReadOnlyScope] 

533 

534 

535class TagList(TagMixin, ListCreateAPI): 

536 """List view for all tags.""" 

537 

538 filterset_class = TagFilter 

539 filter_backends = SEARCH_ORDER_FILTER 

540 ordering_fields = ['name'] 

541 search_fields = ['name'] 

542 

543 

544class TagDetail(TagMixin, RetrieveUpdateDestroyAPI): 

545 """Detail view for a particular tag.""" 

546 

547 

548class CustomUnitViewset(DataExportViewMixin, viewsets.ModelViewSet): 

549 """List view for custom units.""" 

550 

551 queryset = common.models.CustomUnit.objects.all() 

552 serializer_class = common.serializers.CustomUnitSerializer 

553 permission_classes = [IsStaffOrReadOnlyScope] 

554 filter_backends = SEARCH_ORDER_FILTER 

555 

556 @action( 

557 detail=False, 

558 methods=['get'], 

559 serializer_class=common.serializers.AllUnitListResponseSerializer, 

560 ) 

561 def all(self, request, *args, **kwargs): 

562 """Return a list of all available units.""" 

563 reg = InvenTree.conversion.get_unit_registry() 

564 all_units = {k: self.get_unit(reg, k) for k in reg} 

565 data = { 

566 'default_system': reg.default_system, 

567 'available_systems': dir(reg.sys), 

568 'available_units': {k: v for k, v in all_units.items() if v}, 

569 } 

570 return Response(data) 

571 

572 def get_unit(self, reg, k): 

573 """Parse a unit from the registry.""" 

574 if not hasattr(reg, k): 574 ↛ 575line 574 didn't jump to line 575 because the condition on line 574 was never true

575 return None 

576 unit: type[UnitLike] = getattr(reg, k) 

577 return { 

578 'name': k, 

579 'is_alias': reg.get_name(k) == k, 

580 'compatible_units': [str(a) for a in unit.compatible_units()], # ty:ignore[missing-argument] 

581 'isdimensionless': unit.dimensionless, 

582 } 

583 

584 

585common_router.register('units', CustomUnitViewset, basename='api-custom-unit') 

586 

587 

588class ErrorMessageViewSet(BulkDeleteViewsetMixin, RetrieveUpdateDestroyModelViewSet): 

589 """List view for server error messages.""" 

590 

591 queryset = Error.objects.all() 

592 serializer_class = common.serializers.ErrorMessageSerializer 

593 permission_classes = [IsAuthenticatedOrReadScope, IsAdminUser] 

594 

595 filter_backends = SEARCH_ORDER_FILTER 

596 

597 ordering = '-when' 

598 

599 ordering_fields = ['when', 'info'] 

600 

601 search_fields = ['info', 'data'] 

602 

603 

604common_router.register('error-report', ErrorMessageViewSet, basename='api-error') 

605 

606 

607class BackgroundTaskDetail(APIView): 

608 """Detail view for a single background task.""" 

609 

610 permission_classes = [IsAuthenticatedOrReadScope] 

611 

612 @extend_schema(responses={200: common.serializers.TaskDetailSerializer}) 

613 def get(self, request, task_id, *args, **kwargs): 

614 """Fetch information regarding a particular background task ID.""" 

615 response = common.serializers.TaskDetailSerializer.from_task(task_id).data 

616 

617 return Response(response, status=response['http_status']) 

618 

619 

620class BackgroundTaskOverview(APIView): 

621 """Provides an overview of the background task queue status.""" 

622 

623 permission_classes = [IsAuthenticatedOrReadScope, IsAdminUser] 

624 serializer_class = None 

625 

626 @extend_schema( 

627 operation_id='background_task_overview', 

628 responses={200: common.serializers.TaskOverviewSerializer}, 

629 ) 

630 def get(self, request, fmt=None): 

631 """Return information about the current status of the background task queue.""" 

632 import django_q.models as q_models 

633 

634 import InvenTree.status 

635 

636 serializer = common.serializers.TaskOverviewSerializer({ 

637 'is_running': InvenTree.status.is_worker_running(), 

638 'pending_tasks': q_models.OrmQ.objects.count(), 

639 'scheduled_tasks': q_models.Schedule.objects.count(), 

640 'failed_tasks': q_models.Failure.objects.count(), 

641 }) 

642 

643 return Response(serializer.data) 

644 

645 

646class PendingTaskList(BulkDeleteMixin, ListAPI): 

647 """Provides a read-only list of currently pending tasks.""" 

648 

649 permission_classes = [IsAuthenticatedOrReadScope, IsAdminUser] 

650 

651 queryset = django_q.models.OrmQ.objects.all() 

652 serializer_class = common.serializers.PendingTaskSerializer 

653 

654 

655class ScheduledTaskList(ListAPI): 

656 """Provides a read-only list of currently scheduled tasks.""" 

657 

658 permission_classes = [IsAuthenticatedOrReadScope, IsAdminUser] 

659 

660 queryset = django_q.models.Schedule.objects.all() 

661 serializer_class = common.serializers.ScheduledTaskSerializer 

662 

663 filter_backends = SEARCH_ORDER_FILTER 

664 

665 ordering_fields = ['pk', 'func', 'last_run', 'next_run'] 

666 

667 search_fields = ['func', 'name'] 

668 

669 def get_queryset(self): 

670 """Return annotated queryset.""" 

671 queryset = super().get_queryset() 

672 return common.serializers.ScheduledTaskSerializer.annotate_queryset(queryset) 

673 

674 

675class FailedTaskList(BulkDeleteMixin, ListAPI): 

676 """Provides a read-only list of currently failed tasks.""" 

677 

678 permission_classes = [IsAuthenticatedOrReadScope, IsAdminUser] 

679 

680 queryset = django_q.models.Failure.objects.all() 

681 serializer_class = common.serializers.FailedTaskSerializer 

682 

683 filter_backends = SEARCH_ORDER_FILTER 

684 

685 ordering_fields = ['pk', 'func', 'started', 'stopped'] 

686 

687 search_fields = ['func'] 

688 

689 

690class FlagList(ListAPI): 

691 """List view for feature flags.""" 

692 

693 queryset = settings.FLAGS 

694 serializer_class = common.serializers.FlagSerializer 

695 permission_classes = [AllowAnyOrReadScope] 

696 

697 # Specifically disable pagination for this view 

698 pagination_class = None 

699 

700 

701class FlagDetail(RetrieveAPI): 

702 """Detail view for an individual feature flag.""" 

703 

704 serializer_class = common.serializers.FlagSerializer 

705 permission_classes = [AllowAnyOrReadScope] 

706 

707 def get_object(self): 

708 """Attempt to find a config object with the provided key.""" 

709 key = self.kwargs['key'] 

710 value = settings.FLAGS.get(key, None) 

711 if not value: 711 ↛ 712line 711 didn't jump to line 712 because the condition on line 711 was never true

712 raise NotFound() 

713 return {key: value} 

714 

715 

716class ContentTypeList(ListAPI): 

717 """List view for ContentTypes.""" 

718 

719 queryset = ContentType.objects.all() 

720 serializer_class = common.serializers.ContentTypeSerializer 

721 permission_classes = [IsAuthenticatedOrReadScope] 

722 filter_backends = SEARCH_ORDER_FILTER 

723 search_fields = ['app_label', 'model'] 

724 

725 

726class ContentTypeDetail(RetrieveAPI): 

727 """Detail view for a ContentType model.""" 

728 

729 queryset = ContentType.objects.all() 

730 serializer_class = common.serializers.ContentTypeSerializer 

731 permission_classes = [IsAuthenticatedOrReadScope] 

732 

733 

734class ContentTypeModelDetail(ContentTypeDetail): 

735 """Detail view for a ContentType model.""" 

736 

737 def get_object(self): 

738 """Attempt to find a ContentType object with the provided key.""" 

739 model_ref = self.kwargs.get('model', None) 

740 if model_ref: 740 ↛ 746line 740 didn't jump to line 746 because the condition on line 740 was always true

741 qs = self.filter_queryset(self.get_queryset()) 

742 try: 

743 return qs.get(model=model_ref) 

744 except ContentType.DoesNotExist: 

745 raise NotFound() 

746 raise NotFound() 

747 

748 @extend_schema(operation_id='contenttype_retrieve_model') 

749 def get(self, request, *args, **kwargs): 

750 """Detail view for a ContentType model.""" 

751 return super().get(request, *args, **kwargs) 

752 

753 

754class AttachmentFilter(FilterSet): 

755 """Filterset for the AttachmentList API endpoint.""" 

756 

757 class Meta: 

758 """Metaclass options.""" 

759 

760 model = common.models.Attachment 

761 fields = ['model_type', 'model_id', 'upload_user', 'is_image'] 

762 

763 has_thumbnail = rest_filters.BooleanFilter( 

764 label=_('Has Thumbnail'), method='filter_has_thumbnail' 

765 ) 

766 

767 def filter_has_thumbnail(self, queryset, name, value): 

768 """Filter attachments based on whether they have a thumbnail or not.""" 

769 if value: 

770 return queryset.exclude(thumbnail=None).exclude(thumbnail='') 

771 return queryset.filter(Q(thumbnail=None) | Q(thumbnail='')).distinct() 

772 

773 is_link = rest_filters.BooleanFilter(label=_('Is Link'), method='filter_is_link') 

774 

775 def filter_is_link(self, queryset, name, value): 

776 """Filter attachments based on whether they are a link or not.""" 

777 if value: 

778 return queryset.exclude(link=None).exclude(link='') 

779 return queryset.filter(Q(link=None) | Q(link='')).distinct() 

780 

781 is_file = rest_filters.BooleanFilter(label=_('Is File'), method='filter_is_file') 

782 

783 def filter_is_file(self, queryset, name, value): 

784 """Filter attachments based on whether they are a file or not.""" 

785 if value: 

786 return queryset.exclude(attachment=None).exclude(attachment='') 

787 return queryset.filter(Q(attachment=None) | Q(attachment='')).distinct() 

788 

789 tags = common.filters.TagsFilter() 

790 

791 

792class AttachmentMixin: 

793 """Mixin class for Attachment views.""" 

794 

795 queryset = common.models.Attachment.objects.all() 

796 serializer_class = common.serializers.AttachmentSerializer 

797 permission_classes = [IsAuthenticatedOrReadScope] 

798 

799 

800class AttachmentList(AttachmentMixin, BulkDeleteMixin, ListCreateAPI): 

801 """List API endpoint for Attachment objects.""" 

802 

803 filter_backends = SEARCH_ORDER_FILTER 

804 filterset_class = AttachmentFilter 

805 

806 ordering_fields = ['model_id', 'model_type', 'upload_date', 'file_size'] 

807 search_fields = ['comment', 'model_id', 'model_type'] 

808 

809 def perform_create(self, serializer): 

810 """Save the user information when a file is uploaded.""" 

811 serializer.save(upload_user=self.request.user) 

812 

813 def validate_delete(self, queryset, request) -> None: 

814 """Ensure that the user has correct permissions for a bulk-delete. 

815 

816 - Extract all model types from the provided queryset 

817 - Ensure that the user has correct 'delete' permissions for each model 

818 """ 

819 from common.validators import attachment_model_class_from_label 

820 from users.permissions import check_user_permission 

821 

822 model_types = queryset.values_list('model_type', flat=True).distinct() 

823 

824 for model_type in model_types: 

825 if model_class := attachment_model_class_from_label(model_type): 

826 if not check_user_permission(request.user, model_class, 'delete'): 

827 raise ValidationError( 

828 _('User does not have permission to delete these attachments') 

829 ) 

830 

831 

832class AttachmentDetail(AttachmentMixin, RetrieveUpdateDestroyAPI): 

833 """Detail API endpoint for Attachment objects.""" 

834 

835 def update(self, request, *args, **kwargs): 

836 """Update an existing attachment object.""" 

837 attachment = self.get_object() 

838 

839 if not attachment.check_permission('change', request.user): 

840 raise PermissionDenied( 

841 _('User does not have permission to edit this attachment') 

842 ) 

843 

844 partial = kwargs.pop('partial', False) 

845 data = self.clean_data(request.data) 

846 

847 # Extract filename first 

848 filename = data.pop('filename', None) 

849 

850 # Run other validation / updates first, before attempting to rename the file 

851 serializer = self.get_serializer(attachment, data=data, partial=partial) 

852 serializer.is_valid(raise_exception=True) 

853 self.perform_update(serializer) 

854 

855 # User is attempting to rename the file 

856 if filename and attachment.basename and filename != attachment.basename: 

857 attachment.rename(filename) 

858 

859 return Response(serializer.data) 

860 

861 def destroy(self, request, *args, **kwargs): 

862 """Delete an existing attachment object.""" 

863 attachment = self.get_object() 

864 

865 if not attachment.check_permission('delete', request.user): 

866 raise PermissionDenied( 

867 _('User does not have permission to delete this attachment') 

868 ) 

869 

870 return super().destroy(request, *args, **kwargs) 

871 

872 

873class ParameterTemplateFilter(FilterSet): 

874 """FilterSet class for the ParameterTemplateList API endpoint.""" 

875 

876 class Meta: 

877 """Metaclass options.""" 

878 

879 model = common.models.ParameterTemplate 

880 fields = ['name', 'units', 'checkbox', 'enabled'] 

881 

882 has_choices = rest_filters.BooleanFilter( 

883 method='filter_has_choices', label='Has Choice' 

884 ) 

885 

886 def filter_has_choices(self, queryset, name, value): 

887 """Filter queryset to include only PartParameterTemplates with choices.""" 

888 if str2bool(value): 

889 return queryset.exclude(Q(choices=None) | Q(choices='')) 

890 

891 return queryset.filter(Q(choices=None) | Q(choices='')).distinct() 

892 

893 has_units = rest_filters.BooleanFilter(method='filter_has_units', label='Has Units') 

894 

895 def filter_has_units(self, queryset, name, value): 

896 """Filter queryset to include only PartParameterTemplates with units.""" 

897 if str2bool(value): 

898 return queryset.exclude(Q(units=None) | Q(units='')) 

899 

900 return queryset.filter(Q(units=None) | Q(units='')).distinct() 

901 

902 model_type = rest_filters.CharFilter(method='filter_model_type', label='Model Type') 

903 

904 def filter_model_type(self, queryset, name, value): 

905 """Filter queryset to include only ParameterTemplates of the given model type.""" 

906 return common.filters.filter_content_type( 

907 queryset, 'model_type', value, allow_null=False 

908 ) 

909 

910 for_model = rest_filters.CharFilter(method='filter_for_model', label='For Model') 

911 

912 def filter_for_model(self, queryset, name, value): 

913 """Filter queryset to include only ParameterTemplates which apply to the given model. 

914 

915 Note that this varies from the 'model_type' filter, in that ParameterTemplates 

916 with a blank 'model_type' are considered to apply to all models. 

917 """ 

918 return common.filters.filter_content_type( 

919 queryset, 'model_type', value, allow_null=True 

920 ) 

921 

922 exists_for_model = rest_filters.CharFilter( 

923 method='filter_exists_for_model', label='Exists For Model' 

924 ) 

925 

926 def filter_exists_for_model(self, queryset, name, value): 

927 """Filter queryset to include only ParameterTemplates which have at least one Parameter for the given model type.""" 

928 content_type = common.filters.determine_content_type(value) 

929 

930 if not content_type: 930 ↛ 936line 930 didn't jump to line 936 because the condition on line 930 was always true

931 raise ValidationError({ 

932 'exists_for_model': 'Invalid model type provided - unable to determine content type' 

933 }) 

934 

935 # If the 'filter_exists_for_model_id' filter is applied, defer to that 

936 if self.request.query_params.get('exists_for_model_id', None): 

937 return queryset 

938 

939 queryset = queryset.prefetch_related('parameters') 

940 

941 # Annotate the queryset to determine which ParameterTemplates have at least one Parameter for the given model type 

942 queryset = queryset.annotate( 

943 parameter_count=SubqueryCount( 

944 'parameters', filter=Q(model_type=content_type) 

945 ) 

946 ) 

947 

948 # Return only those ParameterTemplates which have at least one Parameter for the given model type 

949 return queryset.filter(parameter_count__gt=0) 

950 

951 exists_for_model_id = rest_filters.NumberFilter( 

952 method='filter_exists_for_model_id', label='Exists For Model ID' 

953 ) 

954 

955 def filter_exists_for_model_id(self, queryset, name, value): 

956 """Filter queryset to include only ParameterTemplates which have at least one Parameter for the given model type and model id. 

957 

958 Notes: 

959 - This filter can only be applied if the 'exists_for_model' filter is also applied, as the model_id is only meaningful in the context of a particular model type. 

960 

961 Reference: https://github.com/inventree/InvenTree/issues/11381 

962 """ 

963 exists_for_model = self.request.query_params.get('exists_for_model', None) 

964 

965 if not exists_for_model: 965 ↛ 970line 965 didn't jump to line 970 because the condition on line 965 was always true

966 raise ValidationError({ 

967 'exists_for_model': 'Invalid model type provided - unable to determine content type' 

968 }) 

969 

970 content_type = common.filters.determine_content_type(exists_for_model) 

971 

972 if not content_type: 

973 raise ValidationError({ 

974 'exists_for_model': 'Invalid model type provided - unable to determine content type' 

975 }) 

976 

977 model_class = content_type.model_class() 

978 

979 # Try to find the model instance 

980 try: 

981 instance = model_class.objects.get(pk=value) 

982 except (model_class.DoesNotExist, ValueError): 

983 # If the model instance does not exist, then we can return an empty queryset 

984 raise ValidationError({ 

985 'exists_for_model_id': 'Invalid model id provided - no such instance for the given model type' 

986 }) 

987 

988 # If the provided model is a "tree" structure, then we should also include any child objects in the filter 

989 if isinstance(instance, InvenTree.models.InvenTreeTree): 

990 id_values = list( 

991 instance.get_descendants(include_self=True).values_list('pk', flat=True) 

992 ) 

993 else: 

994 id_values = [instance.pk] 

995 

996 # Now, filter against model type and model id 

997 queryset = queryset.prefetch_related('parameters') 

998 

999 filters = {'model_type': content_type, 'model_id__in': id_values} 

1000 

1001 # Annotate the queryset to determine which ParameterTemplates have at least one Parameter defined 

1002 queryset = queryset.annotate( 

1003 parameter_count=SubqueryCount('parameters', filter=Q(**filters)) 

1004 ) 

1005 

1006 return queryset.filter(parameter_count__gt=0) 

1007 

1008 exists_for_related_model = rest_filters.CharFilter( 

1009 method='filter_exists_for_related_model', label='Exists For Related Model' 

1010 ) 

1011 

1012 def filter_exists_for_related_model(self, queryset, name, value): 

1013 """Filter applied to map parameter templates to a particular model relation against the target model. 

1014 

1015 For instance, specify 'category' to filter part parameters which exist for any part in that category. 

1016 

1017 Note: 

1018 - This filter has no effect on its own 

1019 - It requires the 'exists_for_model' filter to be applied (to specify the base model) 

1020 - It requires the 'exists_for_related_model_id' filter to be applied also (to specify the related model id) 

1021 """ 

1022 return queryset 

1023 

1024 exists_for_related_model_id = rest_filters.NumberFilter( 

1025 method='filter_exists_for_related_model_id', label='Exists For Model ID' 

1026 ) 

1027 

1028 def filter_exists_for_related_model_id(self, queryset, name, value): 

1029 """Filter queryset to include only ParameterTemplates which have at least one Parameter for the given related model type and model id. 

1030 

1031 Notes: 

1032 - This filter can only be applied if the 'exists_for_model' filter is also applied, as the model_id is only meaningful in the context of a base model 

1033 - This filter can only be applied if the 'exists_for_related_model' filter is also applied, as the related model id is only meaningful in the context of a particular model relation 

1034 

1035 Example: To filter part parameters which have at least one parameter defined for any part in category 5, you could apply the following filters: 

1036 - exists_for_model=part 

1037 - exists_for_related_model=category 

1038 - exists_for_related_model_id=5 

1039 """ 

1040 model = self.request.query_params.get('exists_for_model', None) 

1041 related_model = self.request.query_params.get('exists_for_related_model', None) 

1042 

1043 if not model or not related_model: 

1044 raise ValidationError({ 

1045 'exists_for_model': 'Invalid model type provided - unable to determine content type' 

1046 }) 

1047 

1048 # Determine content type for the base model, to ensure they are valid 

1049 model_type = common.filters.determine_content_type(model) 

1050 

1051 if not model_type: 

1052 return queryset.none() 

1053 

1054 # Determine the model class for the 'related' model 

1055 try: 

1056 related_model_field = model_type.model_class()._meta.get_field( 

1057 related_model 

1058 ) 

1059 except FieldDoesNotExist: 

1060 raise ValidationError({ 

1061 'exists_for_related_model': 'Invalid related model - no such field on the base model' 

1062 }) 

1063 if related_model_field := model_type.model_class()._meta.get_field( 

1064 related_model 

1065 ): 

1066 related_model_class = related_model_field.related_model 

1067 else: 

1068 # Return an empty queryset if the provided related model is invalid 

1069 return queryset.none() 

1070 

1071 # Find all instances of the related model which match the provided related model id 

1072 try: 

1073 related_instance = related_model_class.objects.get(pk=value) 

1074 except (related_model_class.DoesNotExist, ValueError): 

1075 return queryset.none() 

1076 

1077 # Account for potential tree structure in the related model 

1078 if isinstance(related_instance, InvenTree.models.InvenTreeTree): 

1079 related_instances = list( 

1080 related_instance.get_descendants(include_self=True).values_list( 

1081 'pk', flat=True 

1082 ) 

1083 ) 

1084 else: 

1085 related_instances = [related_instance.pk] 

1086 

1087 # Next, find all instances of the base model which are related to the related model instances 

1088 model_instances = model_type.model_class().objects.filter(**{ 

1089 f'{related_model}__in': related_instances 

1090 }) 

1091 model_instance_ids = list(model_instances.values_list('pk', flat=True)) 

1092 

1093 # Now, filter against model type and model id 

1094 queryset = queryset.prefetch_related('parameters') 

1095 

1096 filters = {'model_type': model_type, 'model_id__in': model_instance_ids} 

1097 

1098 # Annotate the queryset to determine which ParameterTemplates have at least one Parameter defined 

1099 queryset = queryset.annotate( 

1100 parameter_count=SubqueryCount('parameters', filter=Q(**filters)) 

1101 ) 

1102 

1103 return queryset.filter(parameter_count__gt=0) 

1104 

1105 

1106class ParameterTemplateMixin: 

1107 """Mixin class for ParameterTemplate views.""" 

1108 

1109 queryset = common.models.ParameterTemplate.objects.all().prefetch_related( 

1110 'model_type' 

1111 ) 

1112 serializer_class = common.serializers.ParameterTemplateSerializer 

1113 permission_classes = [IsAuthenticatedOrReadScope] 

1114 

1115 

1116class ParameterTemplateList(ParameterTemplateMixin, DataExportViewMixin, ListCreateAPI): 

1117 """List view for ParameterTemplate objects.""" 

1118 

1119 filterset_class = ParameterTemplateFilter 

1120 filter_backends = SEARCH_ORDER_FILTER 

1121 search_fields = ['name', 'description'] 

1122 ordering_fields = ['name', 'units', 'checkbox'] 

1123 

1124 

1125class ParameterTemplateDetail(ParameterTemplateMixin, RetrieveUpdateDestroyAPI): 

1126 """Detail view for a ParameterTemplate object.""" 

1127 

1128 

1129class ParameterFilter(FilterSet): 

1130 """Custom filters for the ParameterList API endpoint.""" 

1131 

1132 class Meta: 

1133 """Metaclass options for the filterset.""" 

1134 

1135 model = common.models.Parameter 

1136 fields = ['model_id', 'template', 'updated_by'] 

1137 

1138 enabled = rest_filters.BooleanFilter( 

1139 label='Template Enabled', field_name='template__enabled' 

1140 ) 

1141 

1142 model_type = rest_filters.CharFilter(method='filter_model_type', label='Model Type') 

1143 

1144 def filter_model_type(self, queryset, name, value): 

1145 """Filter queryset to include only Parameters of the given model type.""" 

1146 return common.filters.filter_content_type( 

1147 queryset, 'model_type', value, allow_null=False 

1148 ) 

1149 

1150 

1151class ParameterMixin: 

1152 """Mixin class for Parameter views.""" 

1153 

1154 queryset = common.models.Parameter.objects.all().prefetch_related('model_type') 

1155 serializer_class = common.serializers.ParameterSerializer 

1156 permission_classes = [IsAuthenticatedOrReadScope] 

1157 

1158 

1159class ParameterList( 

1160 OutputOptionsMixin, 

1161 ParameterMixin, 

1162 BulkCreateMixin, 

1163 BulkDeleteMixin, 

1164 DataExportViewMixin, 

1165 ListCreateAPI, 

1166): 

1167 """List API endpoint for Parameter objects.""" 

1168 

1169 filterset_class = ParameterFilter 

1170 filter_backends = SEARCH_ORDER_FILTER 

1171 

1172 ordering_fields = ['name', 'data', 'units', 'template', 'updated', 'updated_by'] 

1173 

1174 ordering_field_aliases = { 

1175 'name': 'template__name', 

1176 'units': 'template__units', 

1177 'data': ['data_numeric', 'data'], 

1178 } 

1179 

1180 search_fields = [ 

1181 'data', 

1182 'template__name', 

1183 'template__description', 

1184 'template__units', 

1185 ] 

1186 

1187 unique_create_fields = ['model_type', 'model_id', 'template'] 

1188 

1189 

1190class ParameterDetail(ParameterMixin, RetrieveUpdateDestroyAPI): 

1191 """Detail API endpoint for Parameter objects.""" 

1192 

1193 

1194@method_decorator(cache_control(public=True, max_age=86400), name='dispatch') 

1195class IconList(ListAPI): 

1196 """List view for available icon packages.""" 

1197 

1198 serializer_class = common.serializers.IconPackageSerializer 

1199 permission_classes = [AllowAnyOrReadScope] 

1200 

1201 def get_queryset(self): 

1202 """Return a list of all available icon packages.""" 

1203 return list(get_icon_packs().values()) 

1204 

1205 

1206class SelectionListMixin(OutputOptionsMixin): 

1207 """Mixin for SelectionList views.""" 

1208 

1209 queryset = common.models.SelectionList.objects.all() 

1210 serializer_class = common.serializers.SelectionListSerializer 

1211 permission_classes = [IsAuthenticatedOrReadScope] 

1212 

1213 def get_queryset(self): 

1214 """Override the queryset method to include entry count.""" 

1215 return self.serializer_class.annotate_queryset(super().get_queryset()) 

1216 

1217 

1218class SelectionListList(SelectionListMixin, ListCreateAPI): 

1219 """List view for SelectionList objects.""" 

1220 

1221 

1222class SelectionListDetail(SelectionListMixin, RetrieveUpdateDestroyAPI): 

1223 """Detail view for a SelectionList object.""" 

1224 

1225 

1226class EntryMixin: 

1227 """Mixin for SelectionEntry views.""" 

1228 

1229 queryset = common.models.SelectionListEntry.objects.all() 

1230 serializer_class = common.serializers.SelectionEntrySerializer 

1231 permission_classes = [IsAuthenticatedOrReadScope] 

1232 lookup_url_kwarg = 'entrypk' 

1233 

1234 def get_queryset(self): 

1235 """Prefetch related fields.""" 

1236 pk = self.kwargs.get('pk', None) 

1237 queryset = super().get_queryset().filter(list=pk) 

1238 queryset = queryset.prefetch_related('list') 

1239 return queryset 

1240 

1241 def perform_destroy(self, instance): 

1242 """Prevent deletion of entries belonging to a locked selection list.""" 

1243 if instance.list.locked: 

1244 raise PermissionDenied(_('Selection list is locked')) 

1245 super().perform_destroy(instance) 

1246 

1247 

1248class SelectionEntryList(EntryMixin, ListCreateAPI): 

1249 """List view for SelectionEntry objects.""" 

1250 

1251 filter_backends = SEARCH_ORDER_FILTER 

1252 

1253 ordering_fields = ['list', 'label', 'active'] 

1254 

1255 search_fields = ['label', 'description'] 

1256 

1257 filterset_fields = ['active', 'value', 'list'] 

1258 

1259 

1260class SelectionEntryDetail(EntryMixin, RetrieveUpdateDestroyAPI): 

1261 """Detail view for a SelectionEntry object.""" 

1262 

1263 

1264class DataOutputViewSet(BulkDeleteViewsetMixin, RetrieveDestroyModelViewSet): 

1265 """Mixin class for DataOutput endpoints.""" 

1266 

1267 queryset = common.models.DataOutput.objects.all() 

1268 serializer_class = common.serializers.DataOutputSerializer 

1269 permission_classes = [IsAuthenticatedOrReadScope] 

1270 

1271 filter_backends = SEARCH_ORDER_FILTER 

1272 ordering_fields = ['pk', 'user', 'plugin', 'output_type', 'created'] 

1273 filterset_fields = ['user'] 

1274 

1275 def get_queryset(self): 

1276 """Return the set of DataOutput objects which the user has permission to view.""" 

1277 queryset = super().get_queryset() 

1278 

1279 try: 

1280 user = self.request.user 

1281 except AttributeError: 

1282 raise PermissionDenied('User information is not available') 

1283 

1284 # Allow staff users access to all DataOutput objects 

1285 if user.is_staff: 1285 ↛ 1289line 1285 didn't jump to line 1289 because the condition on line 1285 was always true

1286 return queryset 

1287 

1288 # All other users are limited to viewing their own DataOutput objects 

1289 return queryset.filter(user=user) 

1290 

1291 

1292common_router.register('data-output', DataOutputViewSet, basename='api-data-output') 

1293 

1294 

1295class EmailViewSet(BulkDeleteViewsetMixin, RetrieveDestroyModelViewSet): 

1296 """Backend E-Mail management for administrative purposes.""" 

1297 

1298 queryset = common.models.EmailMessage.objects.all() 

1299 serializer_class = common.serializers.EmailMessageSerializer 

1300 permission_classes = [IsSuperuserOrSuperScope] 

1301 

1302 filter_backends = SEARCH_ORDER_FILTER 

1303 ordering_fields = [ 

1304 'created', 

1305 'subject', 

1306 'to', 

1307 'sender', 

1308 'status', 

1309 'timestamp', 

1310 'direction', 

1311 ] 

1312 search_fields = [ 

1313 'subject', 

1314 'to', 

1315 'sender', 

1316 'global_id', 

1317 'message_id_key', 

1318 'thread_id_key', 

1319 ] 

1320 

1321 @action( 

1322 detail=False, 

1323 methods=['post'], 

1324 serializer_class=common.serializers.TestEmailSerializer, 

1325 ) 

1326 def test(self, request): 

1327 """Send a test email.""" 

1328 serializer = self.get_serializer(data=request.data) 

1329 serializer.is_valid(raise_exception=True) 

1330 

1331 data = serializer.validated_data 

1332 

1333 delivered, reason = send_email( 

1334 subject='Test email from InvenTree', 

1335 body='This is a test email from InvenTree.', 

1336 recipients=[data['email']], 

1337 ) 

1338 if not delivered: 1338 ↛ 1342line 1338 didn't jump to line 1342 because the condition on line 1338 was always true

1339 raise serializers.ValidationError( 

1340 detail=f'Failed to send test email: "{reason}"' 

1341 ) # pragma: no cover 

1342 return Response(serializer.data) 

1343 

1344 

1345admin_router.register('email', EmailViewSet, basename='api-email') 

1346 

1347 

1348class HealthCheckStatusSerializer(serializers.Serializer): 

1349 """Status of the overall system health.""" 

1350 

1351 status = serializers.ChoiceField( 

1352 help_text='Health status of the InvenTree server', 

1353 choices=['ok', 'loading'], 

1354 read_only=True, 

1355 default='ok', 

1356 ) 

1357 

1358 

1359class HealthCheckView(APIView): 

1360 """Simple JSON endpoint for InvenTree health check. 

1361 

1362 Intended to be used by external services to confirm that the InvenTree server is running. 

1363 """ 

1364 

1365 permission_classes = [AllowAnyOrReadScope] 

1366 

1367 @extend_schema( 

1368 responses={ 

1369 200: OpenApiResponse( 

1370 response=HealthCheckStatusSerializer, 

1371 description='InvenTree server health status', 

1372 ) 

1373 } 

1374 ) 

1375 def get(self, request, *args, **kwargs): 

1376 """Simple health check endpoint for monitoring purposes. 

1377 

1378 Use the root API endpoint for more detailed information (using an authenticated request). 

1379 """ 

1380 status = ( 

1381 InvenTree.ready.isPluginRegistryLoaded() 

1382 if settings.PLUGINS_ENABLED 

1383 else True 

1384 ) 

1385 return JsonResponse( 

1386 {'status': 'ok' if status else 'loading'}, status=200 if status else 503 

1387 ) 

1388 

1389 

1390class ObservabilityEndSerializer(serializers.Serializer): 

1391 """Serializer for observability end endpoint.""" 

1392 

1393 traceid = serializers.CharField( 

1394 help_text='Trace ID to end', max_length=128, required=True 

1395 ) 

1396 service = serializers.CharField( 

1397 help_text='Service name', max_length=128, required=True 

1398 ) 

1399 

1400 

1401class ObservabilityEnd(CreateAPI): 

1402 """Endpoint for observability tools.""" 

1403 

1404 # Note: This endpoint can be called anonymously, as it needs to function before the user is authenticated (e.g. during login) 

1405 permission_classes = [AllowAnyOrReadScope] 

1406 serializer_class = ObservabilityEndSerializer 

1407 

1408 def create(self, request, *args, **kwargs): 

1409 """End a trace in the observability system.""" 

1410 if not settings.TRACING_ENABLED: 1410 ↛ 1413line 1410 didn't jump to line 1413 because the condition on line 1410 was always true

1411 return Response({'status': 'ok'}) 

1412 

1413 data = self.get_serializer(data=request.data) 

1414 data.is_valid(raise_exception=True) 

1415 

1416 traceid = data.validated_data['traceid'] 

1417 # service = data.validated_data['service'] # This will become interesting with frontend observability 

1418 

1419 # End the foreign trend via the low level otel API 

1420 tracer = trace.get_tracer(__name__) 

1421 span_context = trace.SpanContext( 

1422 trace_id=int(traceid, 16), 

1423 span_id=0, 

1424 is_remote=True, 

1425 trace_flags=trace.TraceFlags(0x01), 

1426 trace_state=trace.TraceState(), 

1427 ) 

1428 with tracer.start_span('Ending session') as span: 

1429 span.add_event('Ending external trace') 

1430 span.add_link(span_context) 

1431 

1432 return Response({'status': 'ok'}) 

1433 

1434 

1435selection_urls = [ 

1436 path( 

1437 '<int:pk>/', 

1438 include([ 

1439 # Entries 

1440 path( 

1441 'entry/', 

1442 include([ 

1443 path( 

1444 '<int:entrypk>/', 

1445 include([ 

1446 path( 

1447 '', 

1448 SelectionEntryDetail.as_view(), 

1449 name='api-selectionlistentry-detail', 

1450 ) 

1451 ]), 

1452 ), 

1453 path( 

1454 '', 

1455 SelectionEntryList.as_view(), 

1456 name='api-selectionlistentry-list', 

1457 ), 

1458 ]), 

1459 ), 

1460 path('', SelectionListDetail.as_view(), name='api-selectionlist-detail'), 

1461 ]), 

1462 ), 

1463 path('', SelectionListList.as_view(), name='api-selectionlist-list'), 

1464] 

1465 

1466# API URL patterns 

1467settings_api_urls = [ 

1468 # User settings 

1469 path( 

1470 'user/', 

1471 include([ 

1472 # User Settings Detail 

1473 re_path( 

1474 r'^(?P<key>\w+)/', 

1475 UserSettingsDetail.as_view(), 

1476 name='api-user-setting-detail', 

1477 ), 

1478 # User Settings List 

1479 path('', UserSettingsList.as_view(), name='api-user-setting-list'), 

1480 ]), 

1481 ), 

1482 # Global settings 

1483 path( 

1484 'global/', 

1485 include([ 

1486 # Global Settings Detail 

1487 re_path( 

1488 r'^(?P<key>\w+)/', 

1489 GlobalSettingsDetail.as_view(), 

1490 name='api-global-setting-detail', 

1491 ), 

1492 # Global Settings List 

1493 path('', GlobalSettingsList.as_view(), name='api-global-setting-list'), 

1494 ]), 

1495 ), 

1496] 

1497 

1498common_api_urls = [ 

1499 # Webhooks 

1500 path('webhook/<slug:endpoint>/', WebhookView.as_view(), name='api-webhook'), 

1501 # Uploaded images for notes 

1502 path('notes-image-upload/', NotesImageList.as_view(), name='api-notes-image-list'), 

1503 # Background task information 

1504 path( 

1505 'background-task/', 

1506 include([ 

1507 path('pending/', PendingTaskList.as_view(), name='api-pending-task-list'), 

1508 path( 

1509 'scheduled/', 

1510 ScheduledTaskList.as_view(), 

1511 name='api-scheduled-task-list', 

1512 ), 

1513 path('failed/', FailedTaskList.as_view(), name='api-failed-task-list'), 

1514 path( 

1515 '<str:task_id>/', BackgroundTaskDetail.as_view(), name='api-task-detail' 

1516 ), 

1517 path('', BackgroundTaskOverview.as_view(), name='api-task-overview'), 

1518 ]), 

1519 ), 

1520 # Attachments 

1521 path( 

1522 'attachment/', 

1523 include([ 

1524 path( 

1525 '<int:pk>/', 

1526 include([ 

1527 meta_path(common.models.Attachment), 

1528 path('', AttachmentDetail.as_view(), name='api-attachment-detail'), 

1529 ]), 

1530 ), 

1531 path('', AttachmentList.as_view(), name='api-attachment-list'), 

1532 ]), 

1533 ), 

1534 # Parameters and templates 

1535 path( 

1536 'parameter/', 

1537 include([ 

1538 path( 

1539 'template/', 

1540 include([ 

1541 path( 

1542 '<int:pk>/', 

1543 include([ 

1544 meta_path(common.models.ParameterTemplate), 

1545 path( 

1546 '', 

1547 ParameterTemplateDetail.as_view(), 

1548 name='api-parameter-template-detail', 

1549 ), 

1550 ]), 

1551 ), 

1552 path( 

1553 '', 

1554 ParameterTemplateList.as_view(), 

1555 name='api-parameter-template-list', 

1556 ), 

1557 ]), 

1558 ), 

1559 path( 

1560 '<int:pk>/', 

1561 include([ 

1562 meta_path(common.models.Parameter), 

1563 path('', ParameterDetail.as_view(), name='api-parameter-detail'), 

1564 ]), 

1565 ), 

1566 path('', ParameterList.as_view(), name='api-parameter-list'), 

1567 ]), 

1568 ), 

1569 # Metadata 

1570 path( 

1571 'metadata/', 

1572 include([ 

1573 path( 

1574 '<str:model>/<str:lookup_field>/<str:lookup_value>/', 

1575 GenericMetadataView.as_view(), 

1576 name='api-generic-metadata', 

1577 ), 

1578 path( 

1579 '<str:model>/<int:pk>/', 

1580 SimpleGenericMetadataView.as_view(), 

1581 name='api-generic-metadata', 

1582 ), 

1583 ]), 

1584 ), 

1585 # Project codes 

1586 path( 

1587 'project-code/', 

1588 include([ 

1589 path( 

1590 '<int:pk>/', 

1591 include([ 

1592 meta_path(common.models.ProjectCode), 

1593 path( 

1594 '', ProjectCodeDetail.as_view(), name='api-project-code-detail' 

1595 ), 

1596 ]), 

1597 ), 

1598 path('', ProjectCodeList.as_view(), name='api-project-code-list'), 

1599 ]), 

1600 ), 

1601 # Tags (via django-taggit) 

1602 path( 

1603 'tag/', 

1604 include([ 

1605 path('<int:pk>/', TagDetail.as_view(), name='api-tag-detail'), 

1606 path('', TagList.as_view(), name='api-tag-list'), 

1607 ]), 

1608 ), 

1609 # Flags 

1610 path( 

1611 'flags/', 

1612 include([ 

1613 path('<str:key>/', FlagDetail.as_view(), name='api-flag-detail'), 

1614 path('', FlagList.as_view(), name='api-flag-list'), 

1615 ]), 

1616 ), 

1617 # Status 

1618 path('generic/status/', include(generic_states_api_urls)), 

1619 # Contenttype 

1620 path( 

1621 'contenttype/', 

1622 include([ 

1623 path( 

1624 '<int:pk>/', ContentTypeDetail.as_view(), name='api-contenttype-detail' 

1625 ), 

1626 path( 

1627 'model/<str:model>/', 

1628 ContentTypeModelDetail.as_view(), 

1629 name='api-contenttype-detail-modelname', 

1630 ), 

1631 path('', ContentTypeList.as_view(), name='api-contenttype-list'), 

1632 ]), 

1633 ), 

1634 # Icons 

1635 path('icons/', IconList.as_view(), name='api-icon-list'), 

1636 # Selection lists 

1637 path('selection/', include(selection_urls)), 

1638 # System APIs (related to basic system functions) 

1639 path( 

1640 'system/', 

1641 include([ 

1642 # Health check 

1643 path('health/', HealthCheckView.as_view(), name='api-system-health') 

1644 ]), 

1645 ), 

1646 # Internal System APIs - DO NOT USE 

1647 path( 

1648 'system-internal/', 

1649 include([ 

1650 # Observability 

1651 path( 

1652 'observability/end', 

1653 ObservabilityEnd.as_view(), 

1654 name='api-system-observability', 

1655 ) 

1656 ]), 

1657 ), 

1658 # Router 

1659 path('', include(common_router.urls)), 

1660] 

1661 

1662admin_api_urls = admin_router.urls