4565001079

Build Type

push

github

Committed by Oliver Walters

Commit Message

Use the attribute name as the dict key

Run Details

25 of 25 new or added lines in 2 files covered. (100.0%)

26493 of 30115 relevant lines covered (87.97%)

0.88 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

83.22

/InvenTree/InvenTree/api.py

"""Main JSON interface views."""

from django.conf import settings
from django.db import transaction
from django.http import JsonResponse
from django.utils.translation import gettext_lazy as _

from django_filters.rest_framework import DjangoFilterBackend
from django_q.models import OrmQ
from rest_framework import filters, permissions
from rest_framework.response import Response
from rest_framework.serializers import ValidationError
from rest_framework.views import APIView

import users.models
from InvenTree.mixins import ListCreateAPI
from InvenTree.permissions import RolePermission
from part.templatetags.inventree_extras import plugins_info

from .status import is_worker_running
from .version import (inventreeApiVersion, inventreeInstanceName,
                      inventreeVersion)
from .views import AjaxView


class InfoView(AjaxView):
    """Simple JSON endpoint for InvenTree information.

    Use to confirm that the server is running, etc.
    """

    permission_classes = [permissions.AllowAny]

    def worker_pending_tasks(self):
        """Return the current number of outstanding background tasks"""

        return OrmQ.objects.count()

    def get(self, request, *args, **kwargs):
        """Serve current server information."""
        data = {
            'server': 'InvenTree',
            'version': inventreeVersion(),
            'instance': inventreeInstanceName(),
            'apiVersion': inventreeApiVersion(),
            'worker_running': is_worker_running(),
            'worker_pending_tasks': self.worker_pending_tasks(),
            'plugins_enabled': settings.PLUGINS_ENABLED,
            'active_plugins': plugins_info(),
        }

        return JsonResponse(data)


class NotFoundView(AjaxView):
    """Simple JSON view when accessing an invalid API view."""

    permission_classes = [permissions.AllowAny]

    def get(self, request, *args, **kwargs):
        """Proces an `not found` event on the API."""
        data = {
            'details': _('API endpoint not found'),
            'url': request.build_absolute_uri(),
        }

        return JsonResponse(data, status=404)


class BulkDeleteMixin:
    """Mixin class for enabling 'bulk delete' operations for various models.

    Bulk delete allows for multiple items to be deleted in a single API query,
    rather than using multiple API calls to the various detail endpoints.

    This is implemented for two major reasons:
    - Atomicity (guaranteed that either *all* items are deleted, or *none*)
    - Speed (single API call and DB query)
    """

    def filter_delete_queryset(self, queryset, request):
        """Provide custom filtering for the queryset *before* it is deleted"""
        return queryset

    def delete(self, request, *args, **kwargs):
        """Perform a DELETE operation against this list endpoint.

        We expect a list of primary-key (ID) values to be supplied as a JSON object, e.g.
        {
            items: [4, 8, 15, 16, 23, 42]
        }

        """
        model = self.serializer_class.Meta.model

        # Extract the items from the request body
        try:
            items = request.data.getlist('items', None)
        except AttributeError:
            items = request.data.get('items', None)

        # Extract the filters from the request body
        try:
            filters = request.data.getlist('filters', None)
        except AttributeError:
            filters = request.data.get('filters', None)

        if not items and not filters:
            raise ValidationError({
                "non_field_errors": ["List of items or filters must be provided for bulk deletion"],
            })

        if items and type(items) is not list:
            raise ValidationError({
                "items": ["'items' must be supplied as a list object"]
            })

        if filters and type(filters) is not dict:
            raise ValidationError({
                "filters": ["'filters' must be supplied as a dict object"]
            })

        # Keep track of how many items we deleted
        n_deleted = 0

        with transaction.atomic():

            # Start with *all* models and perform basic filtering
            queryset = model.objects.all()
            queryset = self.filter_delete_queryset(queryset, request)

            # Filter by provided item ID values
            if items:
                queryset = queryset.filter(id__in=items)

            # Filter by provided filters
            if filters:
                queryset = queryset.filter(**filters)

            n_deleted = queryset.count()
            queryset.delete()

        return Response(
            {
                'success': f"Deleted {n_deleted} items",
            },
            status=204
        )


class ListCreateDestroyAPIView(BulkDeleteMixin, ListCreateAPI):
    """Custom API endpoint which provides BulkDelete functionality in addition to List and Create"""
    ...


class APIDownloadMixin:
    """Mixin for enabling a LIST endpoint to be downloaded a file.

    To download the data, add the ?export=<fmt> to the query string.

    The implementing class must provided a download_queryset method,
    e.g.

    def download_queryset(self, queryset, export_format):
        dataset = StockItemResource().export(queryset=queryset)

        filedata = dataset.export(export_format)

        filename = 'InvenTree_Stocktake_{date}.{fmt}'.format(
            date=datetime.now().strftime("%d-%b-%Y"),
            fmt=export_format
        )

        return DownloadFile(filedata, filename)
    """

    def get(self, request, *args, **kwargs):
        """Generic handler for a download request."""
        export_format = request.query_params.get('export', None)

        if export_format and export_format in ['csv', 'tsv', 'xls', 'xlsx']:
            queryset = self.filter_queryset(self.get_queryset())
            return self.download_queryset(queryset, export_format)

        else:
            # Default to the parent class implementation
            return super().get(request, *args, **kwargs)

    def download_queryset(self, queryset, export_format):
        """This function must be implemented to provide a downloadFile request."""
        raise NotImplementedError("download_queryset method not implemented!")


class AttachmentMixin:
    """Mixin for creating attachment objects, and ensuring the user information is saved correctly."""

    permission_classes = [
        permissions.IsAuthenticated,
        RolePermission,
    ]

    filter_backends = [
        DjangoFilterBackend,
        filters.OrderingFilter,
        filters.SearchFilter,
    ]

    def perform_create(self, serializer):
        """Save the user information when a file is uploaded."""
        attachment = serializer.save()
        attachment.user = self.request.user
        attachment.save()


class APISearchView(APIView):
    """A general-purpose 'search' API endpoint

    Returns hits against a number of different models simultaneously,
    to consolidate multiple API requests into a single query.

    Is much more efficient and simplifies code!
    """

    permission_classes = [
        permissions.IsAuthenticated,
    ]

    def get_result_types(self):
        """Construct a list of search types we can return"""

        import build.api
        import company.api
        import order.api
        import part.api
        import stock.api

        return {
            'build': build.api.BuildList,
            'company': company.api.CompanyList,
            'manufacturerpart': company.api.ManufacturerPartList,
            'supplierpart': company.api.SupplierPartList,
            'part': part.api.PartList,
            'partcategory': part.api.CategoryList,
            'purchaseorder': order.api.PurchaseOrderList,
            'returnorder': order.api.ReturnOrderList,
            'salesorder': order.api.SalesOrderList,
            'stockitem': stock.api.StockList,
            'stocklocation': stock.api.StockLocationList,
        }

    def post(self, request, *args, **kwargs):
        """Perform search query against available models"""

        data = request.data

        search = data.get('search', '')

        # Enforce a 'limit' parameter
        try:
            limit = int(data.get('limit', 1))
        except ValueError:
            limit = 1

        try:
            offset = int(data.get('offset', 0))
        except ValueError:
            offset = 0

        results = {}

        for key, cls in self.get_result_types().items():
            # Only return results which are specifically requested
            if key in data:

                params = data[key]

                params['search'] = search

                # Enforce limit
                params['limit'] = limit
                params['offset'] = offset

                # Enforce json encoding
                params['format'] = 'json'

                # Ignore if the params are wrong
                if type(params) is not dict:
                    continue

                view = cls()

                # Override regular query params with specific ones for this search request
                request._request.GET = params
                view.request = request
                view.format_kwarg = 'format'

                # Check permissions and update results dict with particular query
                model = view.serializer_class.Meta.model
                app_label = model._meta.app_label
                model_name = model._meta.model_name
                table = f'{app_label}_{model_name}'

                try:
                    if users.models.RuleSet.check_table_permission(request.user, table, 'view'):
                        results[key] = view.list(request, *args, **kwargs).data
                    else:
                        results[key] = {
                            'error': _('User does not have permission to view this model')
                        }
                except Exception as exc:
                    results[key] = {
                        'error': str(exc)
                    }

        return Response(results)


class StatusView(APIView):
    """Generic API endpoint for discovering information on 'status codes' for a particular model.

    This class should be implemented as a subclass for each type of status.
    For example, the API endpoint /stock/status/ will have information about
    all available 'StockStatus' codes
    """

    permission_classes = [
        permissions.IsAuthenticated,
    ]

    # Override status_class for implementing subclass
    MODEL_REF = 'statusmodel'

    def get_status_model(self, *args, **kwargs):
        """Return the StatusCode moedl based on extra parameters passed to the view"""

        status_model = self.kwargs.get(self.MODEL_REF, None)

        if status_model is None:
            raise ValidationError(f"StatusView view called without '{self.MODEL_REF}' parameter")

        return status_model

    def get(self, request, *args, **kwargs):
        """Perform a GET request to learn information about status codes"""

        status_class = self.get_status_model()

        if not status_class:
            raise NotImplementedError("status_class not defined for this endpoint")

        values = {}

        # Instead of using 'items' directly, we'll look for all 'uppercase' attributes

        for name, value in status_class.names().items():
            entry = {
                'key': value,
                'label': status_class.label(value),
            }

            if hasattr(status_class, 'colors'):
                if color := status_class.colors.get(value, None):
                    entry['color'] = color

            values[name] = entry

        data = {
            'class': status_class.__name__,
            'values': values,
        }

        return Response(data)

1	"""Main JSON interface views."""
2
3	from django.conf import settings	1✔
4	from django.db import transaction	1✔
5	from django.http import JsonResponse	1✔
6	from django.utils.translation import gettext_lazy as _	1✔
7
8	from django_filters.rest_framework import DjangoFilterBackend	1✔
9	from django_q.models import OrmQ	1✔
10	from rest_framework import filters, permissions	1✔
11	from rest_framework.response import Response	1✔
12	from rest_framework.serializers import ValidationError	1✔
13	from rest_framework.views import APIView	1✔
14
15	import users.models	1✔
16	from InvenTree.mixins import ListCreateAPI	1✔
17	from InvenTree.permissions import RolePermission	1✔
18	from part.templatetags.inventree_extras import plugins_info	1✔
19
20	from .status import is_worker_running	1✔
21	from .version import (inventreeApiVersion, inventreeInstanceName,	1✔
22	inventreeVersion)
23	from .views import AjaxView	1✔
24
25
26	class InfoView(AjaxView):	1✔
27	"""Simple JSON endpoint for InvenTree information.
28
29	Use to confirm that the server is running, etc.
30	"""
31
32	permission_classes = [permissions.AllowAny]	1✔
33
34	def worker_pending_tasks(self):	1✔
35	"""Return the current number of outstanding background tasks"""
36
37	return OrmQ.objects.count()	1✔
38
39	def get(self, request, args, *kwargs):	1✔
40	"""Serve current server information."""
41	data = {	1✔
42	'server': 'InvenTree',
43	'version': inventreeVersion(),
44	'instance': inventreeInstanceName(),
45	'apiVersion': inventreeApiVersion(),
46	'worker_running': is_worker_running(),
47	'worker_pending_tasks': self.worker_pending_tasks(),
48	'plugins_enabled': settings.PLUGINS_ENABLED,
49	'active_plugins': plugins_info(),
50	}
51
52	return JsonResponse(data)	1✔
53
54
55	class NotFoundView(AjaxView):	1✔
56	"""Simple JSON view when accessing an invalid API view."""
57
58	permission_classes = [permissions.AllowAny]	1✔
59
60	def get(self, request, args, *kwargs):	1✔
61	"""Proces an `not found` event on the API."""
62	data = {	1✔
63	'details': _('API endpoint not found'),
64	'url': request.build_absolute_uri(),
65	}
66
67	return JsonResponse(data, status=404)	1✔
68
69
70	class BulkDeleteMixin:	1✔
71	"""Mixin class for enabling 'bulk delete' operations for various models.
72
73	Bulk delete allows for multiple items to be deleted in a single API query,
74	rather than using multiple API calls to the various detail endpoints.
75
76	This is implemented for two major reasons:
77	- Atomicity (guaranteed that either all items are deleted, or none)
78	- Speed (single API call and DB query)
79	"""
80
81	def filter_delete_queryset(self, queryset, request):	1✔
82	"""Provide custom filtering for the queryset before it is deleted"""
83	return queryset	1✔
84
85	def delete(self, request, args, *kwargs):	1✔
86	"""Perform a DELETE operation against this list endpoint.
87
88	We expect a list of primary-key (ID) values to be supplied as a JSON object, e.g.
89	{
90	items: [4, 8, 15, 16, 23, 42]
91	}
92
93	"""
94	model = self.serializer_class.Meta.model	1✔
95
96	# Extract the items from the request body
97	try:	1✔
98	items = request.data.getlist('items', None)	1✔
99	except AttributeError:	1✔
100	items = request.data.get('items', None)	1✔
101
102	# Extract the filters from the request body
103	try:	1✔
104	filters = request.data.getlist('filters', None)	1✔
105	except AttributeError:	1✔
106	filters = request.data.get('filters', None)	1✔
107
108	if not items and not filters:	1✔
109	raise ValidationError({	1✔
110	"non_field_errors": ["List of items or filters must be provided for bulk deletion"],
111	})
112
113	if items and type(items) is not list:	1✔
114	raise ValidationError({	1✔
115	"items": ["'items' must be supplied as a list object"]
116	})
117
118	if filters and type(filters) is not dict:	1✔
119	raise ValidationError({	1✔
120	"filters": ["'filters' must be supplied as a dict object"]
121	})
122
123	# Keep track of how many items we deleted
124	n_deleted = 0	1✔
125
126	with transaction.atomic():	1✔
127
128	# Start with all models and perform basic filtering
129	queryset = model.objects.all()	1✔
130	queryset = self.filter_delete_queryset(queryset, request)	1✔
131
132	# Filter by provided item ID values
133	if items:	1✔
134	queryset = queryset.filter(id__in=items)	1✔
135
136	# Filter by provided filters
137	if filters:	1✔
138	queryset = queryset.filter(**filters)	1✔
139
140	n_deleted = queryset.count()	1✔
141	queryset.delete()	1✔
142
143	return Response(	1✔
144	{
145	'success': f"Deleted {n_deleted} items",
146	},
147	status=204
148	)
149
150
151	class ListCreateDestroyAPIView(BulkDeleteMixin, ListCreateAPI):	1✔
152	"""Custom API endpoint which provides BulkDelete functionality in addition to List and Create"""
153	...	1✔
154
155
156	class APIDownloadMixin:	1✔
157	"""Mixin for enabling a LIST endpoint to be downloaded a file.
158
159	To download the data, add the ?export=<fmt> to the query string.
160
161	The implementing class must provided a download_queryset method,
162	e.g.
163
164	def download_queryset(self, queryset, export_format):
165	dataset = StockItemResource().export(queryset=queryset)
166
167	filedata = dataset.export(export_format)
168
169	filename = 'InvenTree_Stocktake_{date}.{fmt}'.format(
170	date=datetime.now().strftime("%d-%b-%Y"),
171	fmt=export_format
172	)
173
174	return DownloadFile(filedata, filename)
175	"""
176
177	def get(self, request, args, *kwargs):	1✔
178	"""Generic handler for a download request."""
179	export_format = request.query_params.get('export', None)	1✔
180
181	if export_format and export_format in ['csv', 'tsv', 'xls', 'xlsx']:	1✔
182	queryset = self.filter_queryset(self.get_queryset())	1✔
183	return self.download_queryset(queryset, export_format)	1✔
184
185	else:
186	# Default to the parent class implementation
187	return super().get(request, args, *kwargs)	1✔
188
189	def download_queryset(self, queryset, export_format):	1✔
190	"""This function must be implemented to provide a downloadFile request."""
191	raise NotImplementedError("download_queryset method not implemented!")	×
192
193
194	class AttachmentMixin:	1✔
195	"""Mixin for creating attachment objects, and ensuring the user information is saved correctly."""
196
197	permission_classes = [	1✔
198	permissions.IsAuthenticated,
199	RolePermission,
200	]
201
202	filter_backends = [	1✔
203	DjangoFilterBackend,
204	filters.OrderingFilter,
205	filters.SearchFilter,
206	]
207
208	def perform_create(self, serializer):	1✔
209	"""Save the user information when a file is uploaded."""
210	attachment = serializer.save()	1✔
211	attachment.user = self.request.user	1✔
212	attachment.save()	1✔
213
214
215	class APISearchView(APIView):	1✔
216	"""A general-purpose 'search' API endpoint
217
218	Returns hits against a number of different models simultaneously,
219	to consolidate multiple API requests into a single query.
220
221	Is much more efficient and simplifies code!
222	"""
223
224	permission_classes = [	1✔
225	permissions.IsAuthenticated,
226	]
227
228	def get_result_types(self):	1✔
229	"""Construct a list of search types we can return"""
230
231	import build.api	1✔
232	import company.api	1✔
233	import order.api	1✔
234	import part.api	1✔
235	import stock.api	1✔
236
237	return {	1✔
238	'build': build.api.BuildList,
239	'company': company.api.CompanyList,
240	'manufacturerpart': company.api.ManufacturerPartList,
241	'supplierpart': company.api.SupplierPartList,
242	'part': part.api.PartList,
243	'partcategory': part.api.CategoryList,
244	'purchaseorder': order.api.PurchaseOrderList,
245	'returnorder': order.api.ReturnOrderList,
246	'salesorder': order.api.SalesOrderList,
247	'stockitem': stock.api.StockList,
248	'stocklocation': stock.api.StockLocationList,
249	}
250
251	def post(self, request, args, *kwargs):	1✔
252	"""Perform search query against available models"""
253
254	data = request.data	1✔
255
256	search = data.get('search', '')	1✔
257
258	# Enforce a 'limit' parameter
259	try:	1✔
260	limit = int(data.get('limit', 1))	1✔
261	except ValueError:	×
262	limit = 1	×
263
264	try:	1✔
265	offset = int(data.get('offset', 0))	1✔
266	except ValueError:	×
267	offset = 0	×
268
269	results = {}	1✔
270
271	for key, cls in self.get_result_types().items():	1✔
272	# Only return results which are specifically requested
273	if key in data:	1✔
274
275	params = data[key]	1✔
276
277	params['search'] = search	1✔
278
279	# Enforce limit
280	params['limit'] = limit	1✔
281	params['offset'] = offset	1✔
282
283	# Enforce json encoding
284	params['format'] = 'json'	1✔
285
286	# Ignore if the params are wrong
287	if type(params) is not dict:	1✔
288	continue	×
289
290	view = cls()	1✔
291
292	# Override regular query params with specific ones for this search request
293	request._request.GET = params	1✔
294	view.request = request	1✔
295	view.format_kwarg = 'format'	1✔
296
297	# Check permissions and update results dict with particular query
298	model = view.serializer_class.Meta.model	1✔
299	app_label = model._meta.app_label	1✔
300	model_name = model._meta.model_name	1✔
301	table = f'{app_label}_{model_name}'	1✔
302
303	try:	1✔
304	if users.models.RuleSet.check_table_permission(request.user, table, 'view'):	1✔
305	results[key] = view.list(request, args, *kwargs).data	1✔
306	else:
307	results[key] = {	1✔
308	'error': _('User does not have permission to view this model')
309	}
310	except Exception as exc:	×
311	results[key] = {	×
312	'error': str(exc)
313	}
314
315	return Response(results)	1✔
316
317
318	class StatusView(APIView):	1✔
319	"""Generic API endpoint for discovering information on 'status codes' for a particular model.
320
321	This class should be implemented as a subclass for each type of status.
322	For example, the API endpoint /stock/status/ will have information about
323	all available 'StockStatus' codes
324	"""
325
326	permission_classes = [	1✔
327	permissions.IsAuthenticated,
328	]
329
330	# Override status_class for implementing subclass
331	MODEL_REF = 'statusmodel'	1✔
332
333	def get_status_model(self, args, *kwargs):	1✔
334	"""Return the StatusCode moedl based on extra parameters passed to the view"""
335
336	status_model = self.kwargs.get(self.MODEL_REF, None)	×
337
338	if status_model is None:	×
339	raise ValidationError(f"StatusView view called without '{self.MODEL_REF}' parameter")	×
340
341	return status_model	×
342
343	def get(self, request, args, *kwargs):	1✔
344	"""Perform a GET request to learn information about status codes"""
345
346	status_class = self.get_status_model()	×
347
348	if not status_class:	×
349	raise NotImplementedError("status_class not defined for this endpoint")	×
350
351	values = {}	×
352
353	# Instead of using 'items' directly, we'll look for all 'uppercase' attributes
354
355	for name, value in status_class.names().items():	×
356	entry = {	×
357	'key': value,
358	'label': status_class.label(value),
359	}
360
361	if hasattr(status_class, 'colors'):	×
362	if color := status_class.colors.get(value, None):	×
363	entry['color'] = color	×
364
365	values[name] = entry	×
366
367	data = {	×
368	'class': status_class.__name__,
369	'values': values,
370	}
371
372	return Response(data)	×

inventree / InvenTree / 4565001079

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous