Coverage for src/backend/InvenTree/plugin/api.py: 85%
223 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 17:47 +0000
1"""API for the plugin app."""
3from typing import Optional
5from django.core.exceptions import ValidationError
6from django.urls import include, path, re_path
7from django.utils.translation import gettext_lazy as _
9import django_filters.rest_framework.filters as rest_filters
10from django_filters.rest_framework import DjangoFilterBackend
11from django_filters.rest_framework.filterset import FilterSet
12from drf_spectacular.utils import extend_schema
13from rest_framework import permissions, status
14from rest_framework.exceptions import NotFound
15from rest_framework.response import Response
16from rest_framework.views import APIView
18import InvenTree.permissions
19import plugin.serializers as PluginSerializers
20from InvenTree.api import meta_path
21from InvenTree.filters import SEARCH_ORDER_FILTER
22from InvenTree.helpers import str2bool
23from InvenTree.mixins import (
24 CreateAPI,
25 ListAPI,
26 RetrieveAPI,
27 RetrieveDestroyAPI,
28 RetrieveUpdateAPI,
29 UpdateAPI,
30)
31from plugin.base.action.api import ActionPluginView
32from plugin.base.barcodes.api import barcode_api_urls
33from plugin.base.locate.api import LocatePluginView
34from plugin.base.supplier.api import supplier_api_urls
35from plugin.base.ui.api import ui_plugins_api_urls
36from plugin.models import PluginConfig, PluginSetting, PluginUserSetting
37from plugin.plugin import InvenTreePlugin
38from plugin.registry import registry
41class PluginFilter(FilterSet):
42 """Filter for the PluginConfig model.
44 Provides custom filtering options for the FilterList API endpoint.
45 """
47 class Meta:
48 """Meta for the filter."""
50 model = PluginConfig
51 fields = ['active']
53 mixin = rest_filters.CharFilter(
54 field_name='mixin', method='filter_mixin', label='Mixin'
55 )
57 def filter_mixin(self, queryset, name, value):
58 """Filter by implement mixin.
60 - A comma-separated list of mixin names can be provided.
61 - Only plugins which implement all of the provided mixins will be returned.
62 """
63 matches = []
64 mixins = [x.strip().lower() for x in value.split(',') if x]
66 for result in queryset:
67 match = True
69 for mixin in mixins: 69 ↛ 74line 69 didn't jump to line 74 because the loop on line 69 didn't complete
70 if mixin not in result.mixins(): 70 ↛ 69line 70 didn't jump to line 69 because the condition on line 70 was always true
71 match = False
72 break
74 if match: 74 ↛ 75line 74 didn't jump to line 75 because the condition on line 74 was never true
75 matches.append(result.pk)
77 return queryset.filter(pk__in=matches)
79 builtin = rest_filters.BooleanFilter(
80 field_name='builtin', label=_('Builtin'), method='filter_builtin'
81 )
83 def filter_builtin(self, queryset, name, value):
84 """Filter by 'builtin' flag."""
85 matches = []
87 for result in queryset:
88 if result.is_builtin() == value: 88 ↛ 87line 88 didn't jump to line 87 because the condition on line 88 was always true
89 matches.append(result.pk)
91 return queryset.filter(pk__in=matches)
93 mandatory = rest_filters.BooleanFilter(
94 field_name='mandatory', label=_('Mandatory'), method='filter_mandatory'
95 )
97 def filter_mandatory(self, queryset, name, value):
98 """Filter by 'mandatory' flag."""
99 from django.conf import settings
101 mandatory_keys = [*registry.MANDATORY_PLUGINS, *settings.PLUGINS_MANDATORY]
103 if str2bool(value):
104 return queryset.filter(key__in=mandatory_keys)
105 else:
106 return queryset.exclude(key__in=mandatory_keys)
108 sample = rest_filters.BooleanFilter(
109 field_name='sample', label=_('Sample'), method='filter_sample'
110 )
112 def filter_sample(self, queryset, name, value):
113 """Filter by 'sample' flag."""
114 matches = []
116 for result in queryset:
117 if result.is_sample() == value: 117 ↛ 118line 117 didn't jump to line 118 because the condition on line 117 was never true
118 matches.append(result.pk)
120 return queryset.filter(pk__in=matches)
122 installed = rest_filters.BooleanFilter(
123 field_name='installed', label=_('Installed'), method='filter_installed'
124 )
126 def filter_installed(self, queryset, name, value):
127 """Filter by 'installed' flag."""
128 matches = []
130 for result in queryset:
131 if result.is_installed() == value: 131 ↛ 130line 131 didn't jump to line 130 because the condition on line 131 was always true
132 matches.append(result.pk)
134 return queryset.filter(pk__in=matches)
137class PluginList(ListAPI):
138 """API endpoint for list of PluginConfig objects.
140 - GET: Return a list of all PluginConfig objects
141 """
143 # Allow any logged in user to read this endpoint
144 # This is necessary to allow certain functionality,
145 # e.g. determining which label printing plugins are available
146 permission_classes = [InvenTree.permissions.IsAuthenticatedOrReadScope]
148 filterset_class = PluginFilter
150 serializer_class = PluginSerializers.PluginConfigSerializer
151 queryset = PluginConfig.objects.all()
153 filter_backends = SEARCH_ORDER_FILTER
155 ordering_fields = ['key', 'name', 'active']
157 ordering = ['-active', 'name', 'key']
159 search_fields = ['key', 'name']
162class PluginDetail(RetrieveDestroyAPI):
163 """API detail endpoint for PluginConfig object.
165 get:
166 Return a single PluginConfig object
168 post:
169 Update a PluginConfig
171 delete:
172 Remove a PluginConfig
173 """
175 queryset = PluginConfig.objects.all()
176 serializer_class = PluginSerializers.PluginConfigSerializer
177 permission_classes = [
178 permissions.IsAuthenticated,
179 InvenTree.permissions.IsSuperuserOrReadOnlyOrScope,
180 ]
181 lookup_field = 'key'
182 lookup_url_kwarg = 'plugin'
184 def delete(self, request, *args, **kwargs):
185 """Handle DELETE request for a PluginConfig instance.
187 We only allow plugin deletion if the plugin is not active.
188 """
189 cfg = self.get_object()
191 if cfg.active: 191 ↛ 196line 191 didn't jump to line 196 because the condition on line 191 was always true
192 raise ValidationError({
193 'detail': _('Plugin cannot be deleted as it is currently active')
194 })
196 return super().delete(request, *args, **kwargs)
199class PluginAdminDetail(RetrieveAPI):
200 """Endpoint for viewing admin integration plugin details.
202 This endpoint is used to view the available admin integration options for a plugin.
203 """
205 queryset = PluginConfig.objects.all()
206 serializer_class = PluginSerializers.PluginAdminDetailSerializer
207 permission_classes = [InvenTree.permissions.IsAdminOrAdminScope]
208 lookup_field = 'key'
209 lookup_url_kwarg = 'plugin'
212class PluginInstall(CreateAPI):
213 """Endpoint for installing a new plugin."""
215 queryset = PluginConfig.objects.none()
216 serializer_class = PluginSerializers.PluginConfigInstallSerializer
217 permission_classes = [InvenTree.permissions.IsSuperuserOrSuperScope]
219 def create(self, request, *args, **kwargs):
220 """Install a plugin via the API."""
221 # Clean up input data
222 data = self.clean_data(request.data)
224 serializer = self.get_serializer(data=data)
225 serializer.is_valid(raise_exception=True)
226 result = self.perform_create(serializer)
227 result['input'] = serializer.data
228 headers = self.get_success_headers(serializer.data)
229 return Response(result, status=status.HTTP_201_CREATED, headers=headers)
231 def perform_create(self, serializer):
232 """Saving the serializer instance performs plugin installation."""
233 return serializer.save()
236class PluginUninstall(UpdateAPI):
237 """Endpoint for uninstalling a single plugin."""
239 queryset = PluginConfig.objects.all()
240 serializer_class = PluginSerializers.PluginUninstallSerializer
241 permission_classes = [InvenTree.permissions.IsSuperuserOrSuperScope]
242 lookup_field = 'key'
243 lookup_url_kwarg = 'plugin'
245 def perform_update(self, serializer):
246 """Uninstall the plugin."""
247 serializer.save()
250class PluginActivate(UpdateAPI):
251 """Endpoint for activating a plugin.
253 - PATCH: Activate a plugin
255 Pass a boolean value for the 'active' field.
256 If not provided, it is assumed to be True,
257 and the plugin will be activated.
258 """
260 queryset = PluginConfig.objects.all()
261 serializer_class = PluginSerializers.PluginActivateSerializer
262 permission_classes = [InvenTree.permissions.IsSuperuserOrSuperScope]
263 lookup_field = 'key'
264 lookup_url_kwarg = 'plugin'
266 def get_object(self):
267 """Returns the object for the view."""
268 if self.request.data.get('pk', None): 268 ↛ 269line 268 didn't jump to line 269 because the condition on line 268 was never true
269 return self.queryset.get(pk=self.request.data.get('pk'))
270 return super().get_object()
272 def perform_update(self, serializer):
273 """Activate the plugin."""
274 serializer.save()
277class PluginReload(CreateAPI):
278 """Endpoint for reloading all plugins."""
280 queryset = PluginConfig.objects.none()
281 serializer_class = PluginSerializers.PluginReloadSerializer
282 permission_classes = [InvenTree.permissions.IsSuperuserOrSuperScope]
284 def perform_create(self, serializer):
285 """Saving the serializer instance performs plugin installation."""
286 return serializer.save()
289class PluginSettingList(ListAPI):
290 """List endpoint for all plugin related settings.
292 - read only
293 - only accessible by staff users
294 """
296 queryset = PluginSetting.objects.all()
297 serializer_class = PluginSerializers.PluginSettingSerializer
299 permission_classes = [
300 permissions.IsAuthenticated,
301 InvenTree.permissions.GlobalSettingsPermissions,
302 ]
304 filter_backends = [DjangoFilterBackend]
306 filterset_fields = ['plugin__active', 'plugin__key']
308 @extend_schema(operation_id='plugins_settings_list_all')
309 def get(self, request, *args, **kwargs):
310 """List endpoint for all plugin related settings.
312 - read only
313 - only accessible by staff users
314 """
315 return super().get(request, *args, **kwargs)
318def check_plugin(
319 plugin_slug: Optional[str], plugin_pk: Optional[int]
320) -> InvenTreePlugin:
321 """Check that a plugin for the provided slug exists and get the config.
323 Args:
324 plugin_slug (str): Slug for plugin.
325 plugin_pk (int): Primary key for plugin.
327 Returns:
328 InvenTreePlugin: The config object for the provided plugin.
330 Raises:
331 NotFound: If plugin is not installed
332 NotFound: If plugin is not correctly registered
333 NotFound: If plugin is not active
334 """
335 # Make sure that a plugin reference is specified
336 if plugin_slug is None and plugin_pk is None: 336 ↛ 337line 336 didn't jump to line 337 because the condition on line 336 was never true
337 raise NotFound(detail='Plugin not specified')
339 # Define filter
340 filters = {}
341 if plugin_slug: 341 ↛ 343line 341 didn't jump to line 343 because the condition on line 341 was always true
342 filters['key'] = plugin_slug
343 elif plugin_pk:
344 filters['pk'] = plugin_pk
345 ref = plugin_slug or plugin_pk
347 # Check that the 'plugin' specified is valid
348 try:
349 plugin_cfg = PluginConfig.objects.filter(**filters).first()
350 except PluginConfig.DoesNotExist:
351 raise NotFound(detail=f"Plugin '{ref}' not installed")
353 if plugin_cfg is None:
354 # This only occurs if the plugin mechanism broke
355 raise NotFound(detail=f"Plugin '{ref}' not installed") # pragma: no cover
357 # Check that the plugin is activated
358 if not plugin_cfg.active: 358 ↛ 359line 358 didn't jump to line 359 because the condition on line 358 was never true
359 raise NotFound(detail=f"Plugin '{ref}' is not active")
361 plugin = plugin_cfg.plugin
363 if not plugin: 363 ↛ 364line 363 didn't jump to line 364 because the condition on line 363 was never true
364 raise NotFound(detail=f"Plugin '{ref}' not installed")
366 return plugin
369class PluginAllSettingList(APIView):
370 """List endpoint for all plugin settings for a specific plugin.
372 - GET: return all settings for a plugin config
373 """
375 permission_classes = [
376 permissions.IsAuthenticated,
377 InvenTree.permissions.GlobalSettingsPermissions,
378 ]
380 @extend_schema(
381 responses={200: PluginSerializers.PluginSettingSerializer(many=True)}
382 )
383 def get(self, request, plugin):
384 """Get all settings for a plugin config."""
385 # look up the plugin
386 plugin = check_plugin(plugin, None)
388 settings = getattr(plugin, 'settings', {})
390 settings_dict = PluginSetting.all_settings(
391 settings_definition=settings, plugin=plugin.plugin_config()
392 )
394 results = PluginSerializers.PluginSettingSerializer(
395 list(settings_dict.values()), many=True
396 ).data
397 return Response(results)
400class PluginSettingDetail(RetrieveUpdateAPI):
401 """Detail endpoint for a plugin-specific setting."""
403 queryset = PluginSetting.objects.all()
404 serializer_class = PluginSerializers.PluginSettingSerializer
406 permission_classes = [
407 permissions.IsAuthenticated,
408 InvenTree.permissions.GlobalSettingsPermissions,
409 ]
411 def get_object(self):
412 """Lookup the plugin setting object, based on the URL.
414 The URL provides the 'slug' of the plugin, and the 'key' of the setting.
415 Both the 'slug' and 'key' must be valid, else a 404 error is raised
416 """
417 setting_key = self.kwargs['key']
419 # Look up plugin
420 plugin = check_plugin(self.kwargs.get('plugin', None), None)
422 settings = getattr(plugin, 'settings', {})
424 if setting_key not in settings: 424 ↛ 429line 424 didn't jump to line 429 because the condition on line 424 was always true
425 raise NotFound(
426 detail=f"Plugin '{plugin.slug}' has no setting matching '{setting_key}'"
427 )
429 return PluginSetting.get_setting_object(
430 setting_key, plugin=plugin.plugin_config()
431 )
434class PluginUserSettingList(APIView):
435 """List endpoint for all user settings for a specific plugin.
437 - GET: return all user settings for a plugin config
438 """
440 queryset = PluginUserSetting.objects.all()
441 serializer_class = PluginSerializers.PluginUserSettingSerializer
442 permission_classes = [InvenTree.permissions.UserSettingsPermissionsOrScope]
444 @extend_schema(
445 responses={200: PluginSerializers.PluginUserSettingSerializer(many=True)}
446 )
447 def get(self, request, plugin):
448 """Get all user settings for a plugin config."""
449 # look up the plugin
450 plugin = check_plugin(plugin, None)
452 user_settings = getattr(plugin, 'user_settings', {})
454 settings_dict = PluginUserSetting.all_settings(
455 settings_definition=user_settings,
456 plugin=plugin.plugin_config(),
457 user=request.user,
458 )
460 results = PluginSerializers.PluginUserSettingSerializer(
461 list(settings_dict.values()), many=True
462 ).data
463 return Response(results)
466class PluginUserSettingDetail(RetrieveUpdateAPI):
467 """Detail endpoint for a plugin-specific user setting."""
469 lookup_field = 'key'
470 queryset = PluginUserSetting.objects.all()
471 serializer_class = PluginSerializers.PluginUserSettingSerializer
472 permission_classes = [InvenTree.permissions.UserSettingsPermissionsOrScope]
474 def get_object(self):
475 """Lookup the plugin user setting object, based on the URL."""
476 setting_key = self.kwargs['key']
478 # Look up plugin
479 plugin = check_plugin(self.kwargs.get('plugin', None), None)
481 settings = getattr(plugin, 'user_settings', {})
483 if setting_key not in settings: 483 ↛ 488line 483 didn't jump to line 488 because the condition on line 483 was always true
484 raise NotFound(
485 detail=f"Plugin '{plugin.slug}' has no user setting matching '{setting_key}'"
486 )
488 return PluginUserSetting.get_setting_object(
489 setting_key, plugin=plugin.plugin_config(), user=self.request.user
490 )
493class RegistryStatusView(APIView):
494 """Status API endpoint for the plugin registry.
496 - GET: Provide status data for the plugin registry
497 """
499 permission_classes = [InvenTree.permissions.IsSuperuserOrSuperScope]
501 serializer_class = PluginSerializers.PluginRegistryStatusSerializer
503 @extend_schema(responses={200: PluginSerializers.PluginRegistryStatusSerializer()})
504 def get(self, request):
505 """Show plugin registry status information."""
506 error_list = []
508 for stage, errors in registry.errors.items(): 508 ↛ 509line 508 didn't jump to line 509 because the loop on line 508 never started
509 for error_detail in errors:
510 for name, message in error_detail.items():
511 error_list.append({
512 'stage': stage,
513 'name': name,
514 'message': message,
515 })
517 result = PluginSerializers.PluginRegistryStatusSerializer({
518 'registry_errors': error_list,
519 'active_plugins': PluginConfig.objects.filter(active=True).count(),
520 }).data
522 return Response(result)
525# class PluginMetadataView(MetadataView):
526# """Metadata API endpoint for the PluginConfig model."""
528# lookup_field = 'key'
529# lookup_url_kwarg = 'plugin'
532plugin_api_urls = [
533 path('action/', ActionPluginView.as_view(), name='api-action-plugin'),
534 path('barcode/', include(barcode_api_urls)),
535 path('locate/', LocatePluginView.as_view(), name='api-locate-plugin'),
536 path(
537 'plugins/',
538 include([
539 # UI plugins
540 path('ui/', include(ui_plugins_api_urls)),
541 # Plugin management
542 path('reload/', PluginReload.as_view(), name='api-plugin-reload'),
543 path('install/', PluginInstall.as_view(), name='api-plugin-install'),
544 # Registry status
545 path(
546 'status/',
547 RegistryStatusView.as_view(),
548 name='api-plugin-registry-status',
549 ),
550 path(
551 'settings/',
552 include([
553 path(
554 '', PluginSettingList.as_view(), name='api-plugin-setting-list'
555 )
556 ]),
557 ),
558 # Lookup for individual plugins (based on 'plugin', not 'pk')
559 path(
560 '<str:plugin>/',
561 include([
562 path(
563 'user-settings/',
564 include([
565 re_path(
566 r'^(?P<key>\w+)/',
567 PluginUserSettingDetail.as_view(),
568 name='api-plugin-user-setting-detail',
569 ),
570 path(
571 '',
572 PluginUserSettingList.as_view(),
573 name='api-plugin-user-setting-list',
574 ),
575 ]),
576 ),
577 path(
578 'settings/',
579 include([
580 re_path(
581 r'^(?P<key>\w+)/',
582 PluginSettingDetail.as_view(),
583 name='api-plugin-setting-detail',
584 ),
585 path(
586 '',
587 PluginAllSettingList.as_view(),
588 name='api-plugin-settings',
589 ),
590 ]),
591 ),
592 meta_path(
593 PluginConfig, lookup_field='key', lookup_field_ref='plugin'
594 ),
595 path(
596 'activate/',
597 PluginActivate.as_view(),
598 name='api-plugin-detail-activate',
599 ),
600 path(
601 'uninstall/',
602 PluginUninstall.as_view(),
603 name='api-plugin-uninstall',
604 ),
605 path(
606 'admin/', PluginAdminDetail.as_view(), name='api-plugin-admin'
607 ),
608 path('', PluginDetail.as_view(), name='api-plugin-detail'),
609 ]),
610 ),
611 path('', PluginList.as_view(), name='api-plugin-list'),
612 ]),
613 ),
614 path('supplier/', include(supplier_api_urls)),
615]