Coverage for src/backend/InvenTree/plugin/base/supplier/api.py: 47%
101 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 17:47 +0000
1"""API views for supplier plugins in InvenTree."""
3from typing import TYPE_CHECKING
5from django.db import transaction
6from django.urls import path
8from drf_spectacular.utils import OpenApiParameter, extend_schema
9from rest_framework import status
10from rest_framework.exceptions import NotFound
11from rest_framework.response import Response
12from rest_framework.views import APIView
14from InvenTree import permissions
15from part.models import PartCategoryParameterTemplate
16from plugin import registry
17from plugin.plugin import PluginMixinEnum
19from .serializers import (
20 ImportRequestSerializer,
21 ImportResultSerializer,
22 SearchResultSerializer,
23 SupplierListSerializer,
24)
26if TYPE_CHECKING: 26 ↛ 27line 26 didn't jump to line 27 because the condition on line 26 was never true
27 from plugin.base.supplier.mixins import SupplierMixin
28else: # pragma: no cover
30 class SupplierMixin:
31 """Dummy class for type checking."""
34def get_supplier_plugin(plugin_slug: str, supplier_slug: str) -> SupplierMixin:
35 """Return the supplier plugin for the given plugin and supplier slugs."""
36 supplier_plugin = None
37 for plugin in registry.with_mixin(PluginMixinEnum.SUPPLIER): 37 ↛ 38line 37 didn't jump to line 38 because the loop on line 37 never started
38 if plugin.slug == plugin_slug:
39 supplier_plugin = plugin
40 break
42 if not supplier_plugin: 42 ↛ 45line 42 didn't jump to line 45 because the condition on line 42 was always true
43 raise NotFound(detail=f"Plugin '{plugin_slug}' not found")
45 if not any(s.slug == supplier_slug for s in supplier_plugin.get_suppliers()):
46 raise NotFound(
47 detail=f"Supplier '{supplier_slug}' not found for plugin '{plugin_slug}'"
48 )
50 return supplier_plugin
53class ListSupplier(APIView):
54 """List all available supplier plugins.
56 - GET: List supplier plugins
57 """
59 role_required = 'part.add'
60 permission_classes = [
61 permissions.IsAuthenticatedOrReadScope,
62 permissions.RolePermission,
63 ]
64 serializer_class = SupplierListSerializer
66 @extend_schema(responses={200: SupplierListSerializer(many=True)})
67 def get(self, request):
68 """List all available supplier plugins."""
69 suppliers = []
70 for plugin in registry.with_mixin(PluginMixinEnum.SUPPLIER): 70 ↛ 71line 70 didn't jump to line 71 because the loop on line 70 never started
71 suppliers.extend([
72 {
73 'plugin_slug': plugin.slug,
74 'supplier_slug': supplier.slug,
75 'supplier_name': supplier.name,
76 }
77 for supplier in plugin.get_suppliers()
78 ])
80 return Response(suppliers)
83class SearchPart(APIView):
84 """Search parts by supplier.
86 - GET: Start part search
87 """
89 role_required = 'part.add'
90 permission_classes = [
91 permissions.IsAuthenticatedOrReadScope,
92 permissions.RolePermission,
93 ]
94 serializer_class = SearchResultSerializer
96 @extend_schema(
97 parameters=[
98 OpenApiParameter(name='plugin', description='Plugin slug', required=True),
99 OpenApiParameter(
100 name='supplier', description='Supplier slug', required=True
101 ),
102 OpenApiParameter(name='term', description='Search term', required=True),
103 ],
104 responses={200: SearchResultSerializer(many=True)},
105 )
106 def get(self, request):
107 """Search parts by supplier."""
108 plugin_slug = request.query_params.get('plugin', '')
109 supplier_slug = request.query_params.get('supplier', '')
110 term = request.query_params.get('term', '')
112 supplier_plugin = get_supplier_plugin(plugin_slug, supplier_slug)
113 try:
114 results = supplier_plugin.get_search_results(supplier_slug, term)
115 except Exception as e:
116 return Response(
117 {'error': str(e)}, status=status.HTTP_500_INTERNAL_SERVER_ERROR
118 )
120 response = SearchResultSerializer(results, many=True).data
121 return Response(response)
124class ImportPart(APIView):
125 """Import a part by supplier.
127 - POST: Attempt to import part by sku
128 """
130 role_required = 'part.add'
131 permission_classes = [
132 permissions.IsAuthenticatedOrReadScope,
133 permissions.RolePermission,
134 ]
135 serializer_class = ImportResultSerializer
137 @extend_schema(
138 request=ImportRequestSerializer, responses={200: ImportResultSerializer}
139 )
140 def post(self, request):
141 """Import a part by supplier."""
142 serializer = ImportRequestSerializer(data=request.data)
143 if not serializer.is_valid():
144 return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
146 # Extract validated data
147 plugin_slug = serializer.validated_data.get('plugin', '')
148 supplier_slug = serializer.validated_data.get('supplier', '')
149 part_import_id = serializer.validated_data.get('part_import_id', '')
150 category = serializer.validated_data.get('category_id', None)
151 part = serializer.validated_data.get('part_id', None)
153 supplier_plugin = get_supplier_plugin(plugin_slug, supplier_slug)
155 # Validate part/category
156 if not part and not category:
157 return Response(
158 {
159 'detail': "'category_id' is not provided, but required if no part_id is provided"
160 },
161 status=status.HTTP_400_BAD_REQUEST,
162 )
164 from plugin.base.supplier.mixins import supplier
166 # Import part data
167 try:
168 import_data = supplier_plugin.get_import_data(supplier_slug, part_import_id)
170 with transaction.atomic():
171 # create part if it does not exist
172 if not part:
173 part = supplier_plugin.import_part(
174 import_data, category=category, creation_user=request.user
175 )
177 # create manufacturer part
178 manufacturer_part = supplier_plugin.import_manufacturer_part(
179 import_data, part=part
180 )
182 # create supplier part
183 supplier_part = supplier_plugin.import_supplier_part(
184 import_data, part=part, manufacturer_part=manufacturer_part
185 )
187 # Set as primary supplier if not already set
188 if not part.default_supplier:
189 supplier_part.primary = True
190 supplier_part.save()
192 # get pricing
193 pricing = supplier_plugin.get_pricing_data(import_data)
195 # get parameters
196 parameters = supplier_plugin.get_parameters(import_data)
197 except supplier.PartNotFoundError:
198 return Response(
199 {'detail': f"Part with id: '{part_import_id}' not found"},
200 status=status.HTTP_404_NOT_FOUND,
201 )
202 except Exception as e:
203 return Response(
204 {'detail': str(e)}, status=status.HTTP_500_INTERNAL_SERVER_ERROR
205 )
207 # add default parameters for category
208 if category:
209 categories = category.get_ancestors(include_self=True)
210 category_parameters = PartCategoryParameterTemplate.objects.filter(
211 category__in=categories
212 )
214 for c in category_parameters:
215 for p in parameters:
216 if p.parameter_template == c.template:
217 p.on_category = True
218 p.value = p.value if p.value is not None else c.default_value
219 break
220 else:
221 parameters.append(
222 supplier.ImportParameter(
223 name=c.template.name,
224 value=c.default_value,
225 on_category=True,
226 parameter_template=c.template,
227 )
228 )
229 parameters.sort(key=lambda x: x.on_category, reverse=True)
231 response = ImportResultSerializer({
232 'part_id': part.pk,
233 'part_detail': part,
234 'supplier_part_id': supplier_part.pk,
235 'manufacturer_part_id': manufacturer_part.pk,
236 'pricing': pricing,
237 'parameters': parameters,
238 }).data
239 return Response(response)
242supplier_api_urls = [
243 path('list/', ListSupplier.as_view(), name='api-supplier-list'),
244 path('search/', SearchPart.as_view(), name='api-supplier-search'),
245 path('import/', ImportPart.as_view(), name='api-supplier-import'),
246]