Coverage for src/backend/InvenTree/importer/api.py: 77%
107 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 17:47 +0000
1"""API endpoints for the importer app."""
3from django.shortcuts import get_object_or_404
4from django.urls import include, path
6from drf_spectacular.utils import extend_schema
7from rest_framework import serializers
8from rest_framework.exceptions import PermissionDenied
9from rest_framework.response import Response
10from rest_framework.views import APIView
12import importer.models
13import importer.registry
14import importer.serializers
15import InvenTree.permissions
16from InvenTree.api import BulkDeleteMixin
17from InvenTree.filters import SEARCH_ORDER_FILTER
18from InvenTree.mixins import (
19 CreateAPI,
20 ListAPI,
21 ListCreateAPI,
22 RetrieveUpdateAPI,
23 RetrieveUpdateDestroyAPI,
24)
25from users.permissions import check_user_permission
28class DataImporterPermissionMixin:
29 """Mixin class for checking permissions on DataImporter objects."""
31 # Default permissions: User must be authenticated
32 permission_classes = [
33 InvenTree.permissions.IsAuthenticatedOrReadScope,
34 InvenTree.permissions.DataImporterPermission,
35 ]
38class DataImporterModelSerializer(serializers.Serializer):
39 """Model references to map info that might get imported."""
41 serializer = serializers.CharField(read_only=True)
42 model_type = serializers.CharField(read_only=True)
43 api_url = serializers.URLField(read_only=True, allow_null=True)
46class DataImporterModelList(APIView):
47 """API endpoint for displaying a list of models available for import."""
49 permission_classes = [InvenTree.permissions.IsAuthenticatedOrReadScope]
50 serializer_class = DataImporterModelSerializer(many=True)
52 def get(self, request):
53 """Return a list of models available for import."""
54 models = []
56 for serializer in importer.registry.get_supported_serializers():
57 model = serializer.Meta.model
58 url = model.get_api_url() if hasattr(model, 'get_api_url') else None
60 models.append({
61 'serializer': str(serializer.__name__),
62 'model_type': model.__name__.lower(),
63 'api_url': url,
64 })
66 return Response(models)
69class DataImportSessionMixin:
70 """Mixin class for DataImportSession API views."""
72 queryset = importer.models.DataImportSession.objects.all()
73 serializer_class = importer.serializers.DataImportSessionSerializer
74 permission_classes = [InvenTree.permissions.DataImporterPermission]
76 def get_queryset(self):
77 """Return the set of DataImportSession objects that the user has permission to view."""
78 queryset = super().get_queryset()
80 try:
81 user = self.request.user
82 except AttributeError:
83 raise PermissionDenied('User information is not available')
85 # Allow staff users access to all DataImportSession objects
86 if user.is_staff: 86 ↛ 90line 86 didn't jump to line 90 because the condition on line 86 was always true
87 return queryset
89 # For non-staff users, only allow access to sessions that they have created
90 return queryset.filter(user=user)
93class DataImportSessionList(BulkDeleteMixin, DataImportSessionMixin, ListCreateAPI):
94 """API endpoint for accessing a list of DataImportSession objects."""
96 filter_backends = SEARCH_ORDER_FILTER
97 filterset_fields = ['model_type', 'status', 'user']
98 ordering_fields = ['timestamp', 'status', 'model_type']
101class DataImportSessionDetail(DataImportSessionMixin, RetrieveUpdateDestroyAPI):
102 """Detail endpoint for a single DataImportSession object."""
105class DataImportSessionAcceptFields(APIView):
106 """API endpoint to accept the field mapping for a DataImportSession."""
108 permission_classes = [InvenTree.permissions.IsAuthenticatedOrReadScope]
109 serializer_class = None
111 @extend_schema(
112 responses={200: importer.serializers.DataImportSessionSerializer(many=False)}
113 )
114 def post(self, request, pk):
115 """Accept the field mapping for a DataImportSession."""
116 session = get_object_or_404(importer.models.DataImportSession, pk=pk)
118 # Check session ownership
119 if not request.user.is_staff and session.user != request.user:
120 raise PermissionDenied()
122 # Check that the user has permission to accept the field mapping
123 if model_class := session.model_class:
124 if not check_user_permission(request.user, model_class, 'change'):
125 raise PermissionDenied()
127 # Attempt to accept the mapping (may raise an exception if the mapping is invalid)
128 session.accept_mapping()
130 return Response(importer.serializers.DataImportSessionSerializer(session).data)
133class DataImportSessionAcceptRows(DataImporterPermissionMixin, CreateAPI):
134 """API endpoint to accept the rows for a DataImportSession."""
136 queryset = importer.models.DataImportSession.objects.all()
137 serializer_class = importer.serializers.DataImportAcceptRowSerializer
139 def get_serializer_context(self):
140 """Add the import session object to the serializer context."""
141 ctx = super().get_serializer_context()
143 try:
144 session = importer.models.DataImportSession.objects.get(
145 pk=self.kwargs.get('pk', None)
146 )
147 except importer.models.DataImportSession.DoesNotExist:
148 session = None
150 if session: 150 ↛ 151line 150 didn't jump to line 151 because the condition on line 150 was never true
151 user = self.request.user
152 if not user.is_staff and session.user != user:
153 raise PermissionDenied()
154 ctx['session'] = session
156 ctx['request'] = self.request
157 return ctx
160class DataImportSessionChildMixin(DataImporterPermissionMixin):
161 """Mixin for DataImportRow and DataImportColumnMap views.
163 Ensures users can only access objects that belong to an import session they own.
164 Staff users retain access to all objects.
165 """
167 def get_queryset(self):
168 """Return only objects whose session belongs to the requesting user."""
169 queryset = super().get_queryset()
171 try:
172 user = self.request.user
173 except AttributeError:
174 raise PermissionDenied('User information is not available')
176 if user.is_staff: 176 ↛ 179line 176 didn't jump to line 179 because the condition on line 176 was always true
177 return queryset
179 return queryset.filter(session__user=user)
182class DataImportColumnMappingList(DataImportSessionChildMixin, ListAPI):
183 """API endpoint for accessing a list of DataImportColumnMap objects."""
185 queryset = importer.models.DataImportColumnMap.objects.all()
186 serializer_class = importer.serializers.DataImportColumnMapSerializer
188 filter_backends = SEARCH_ORDER_FILTER
190 filterset_fields = ['session']
193class DataImportColumnMappingDetail(DataImportSessionChildMixin, RetrieveUpdateAPI):
194 """Detail endpoint for a single DataImportColumnMap object."""
196 queryset = importer.models.DataImportColumnMap.objects.all()
197 serializer_class = importer.serializers.DataImportColumnMapSerializer
200class DataImportRowList(DataImportSessionChildMixin, BulkDeleteMixin, ListAPI):
201 """API endpoint for accessing a list of DataImportRow objects."""
203 queryset = importer.models.DataImportRow.objects.all()
204 serializer_class = importer.serializers.DataImportRowSerializer
206 filter_backends = SEARCH_ORDER_FILTER
208 filterset_fields = ['session', 'valid', 'complete']
210 ordering_fields = ['pk', 'row_index', 'valid']
212 ordering = 'row_index'
215class DataImportRowDetail(DataImportSessionChildMixin, RetrieveUpdateDestroyAPI):
216 """Detail endpoint for a single DataImportRow object."""
218 queryset = importer.models.DataImportRow.objects.all()
219 serializer_class = importer.serializers.DataImportRowSerializer
222importer_api_urls = [
223 path('models/', DataImporterModelList.as_view(), name='api-importer-model-list'),
224 path(
225 'session/',
226 include([
227 path(
228 '<int:pk>/',
229 include([
230 path(
231 'accept_fields/',
232 DataImportSessionAcceptFields.as_view(),
233 name='api-import-session-accept-fields',
234 ),
235 path(
236 'accept_rows/',
237 DataImportSessionAcceptRows.as_view(),
238 name='api-import-session-accept-rows',
239 ),
240 path(
241 '',
242 DataImportSessionDetail.as_view(),
243 name='api-import-session-detail',
244 ),
245 ]),
246 ),
247 path('', DataImportSessionList.as_view(), name='api-importer-session-list'),
248 ]),
249 ),
250 path(
251 'column-mapping/',
252 include([
253 path(
254 '<int:pk>/',
255 DataImportColumnMappingDetail.as_view(),
256 name='api-importer-mapping-detail',
257 ),
258 path(
259 '',
260 DataImportColumnMappingList.as_view(),
261 name='api-importer-mapping-list',
262 ),
263 ]),
264 ),
265 path(
266 'row/',
267 include([
268 path(
269 '<int:pk>/',
270 DataImportRowDetail.as_view(),
271 name='api-importer-row-detail',
272 ),
273 path('', DataImportRowList.as_view(), name='api-importer-row-list'),
274 ]),
275 ),
276]