from django.shortcuts import render from django.contrib.auth.models import User, Group from django.core.exceptions import ObjectDoesNotExist from rest_framework import viewsets, status from rest_framework import permissions from rest_framework.views import APIView from .serializers import * from parts import models as parts_models from rest_framework.response import Response from django.db.models.deletion import ProtectedError from django.db.models import F import datetime from datetime import timedelta from django.conf import settings from django.utils import timezone from rest_framework.authtoken.views import ObtainAuthToken from rest_framework.authtoken.models import Token from rest_framework.throttling import AnonRateThrottle from rest_framework.decorators import action from rest_framework import filters import django_filters.rest_framework # Create your views here. class UserViewSet(viewsets.ReadOnlyModelViewSet): """ API endpoint that allows users to be viewed or edited. """ queryset = User.objects.all() serializer_class = UserSerializer permission_classes = [permissions.IsAuthenticated] filter_backends = [filters.SearchFilter] search_fields = ['username', 'first_name', 'last_name', 'email'] class GroupViewSet(viewsets.ReadOnlyModelViewSet): """ API endpoint that allows users to be viewed or edited. """ queryset = Group.objects.all() serializer_class = GroupSerializer permission_classes = [permissions.IsAuthenticated] class PartsStorageViewSet(viewsets.ModelViewSet): queryset = parts_models.Storage.objects.all() serializer_class = StorageSerializer permission_classes = [permissions.DjangoModelPermissions] filter_backends = [django_filters.rest_framework.DjangoFilterBackend] filterset_fields = ['id', 'name', 'parent_storage'] class PartsComponentViewSet(viewsets.ModelViewSet): queryset = parts_models.Component.objects.all() serializer_class = ComponentSerializer permission_classes = [permissions.DjangoModelPermissions] filter_backends = [filters.SearchFilter] search_fields = ['id', 'name', 'package__name', 'manufacturer__name'] class PartsComponentTypeViewSet(viewsets.ModelViewSet): queryset = parts_models.ComponentType.objects.all() serializer_class = ComponentTypeSerializer permission_classes = [permissions.DjangoModelPermissions] filter_backends = [filters.SearchFilter] search_fields = ['class_name'] class PartsComponentParameterTypeViewSet(viewsets.ModelViewSet): queryset = parts_models.ComponentParameterType.objects.all() serializer_class = ComponentParameterTypeSerializer permission_classes = [permissions.DjangoModelPermissions] filter_backends = [filters.SearchFilter] search_fields = ['parameter_name', 'unit'] class PartsManufacturerViewSet(viewsets.ModelViewSet): queryset = parts_models.Manufacturer.objects.all() serializer_class = ManufacturerSerializer permission_classes = [permissions.DjangoModelPermissions] filter_backends = [filters.SearchFilter] search_fields = ['name'] class PartsStockViewSet(viewsets.ModelViewSet): queryset = parts_models.Stock.objects.all() serializer_class = StockSerializer permission_classes = [permissions.DjangoModelPermissions] @action(detail=True, methods=['patch'], name="change-stock-count") def update_stock(self, request, pk=None): stock = self.get_object() serializer = StockIncrementDecrementSerializer(data=request.data) if serializer.is_valid(): increment = serializer.data['increment'] if stock.atomic_increment(increment): return Response({'status': 'Stock updated', 'update_value': increment}) else: return Response({'status': 'Stock not updated. Would be negative'}, status=status.HTTP_400_BAD_REQUEST) else: return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) class PartsPackageViewSet(viewsets.ModelViewSet): queryset = parts_models.Package.objects.all() serializer_class = PackageSerializer permission_classes = [permissions.DjangoModelPermissions] filter_backends = [filters.SearchFilter] search_fields = ['name'] class PartsDistributorviewSet(viewsets.ModelViewSet): queryset = parts_models.Distributor.objects.all() serializer_class = DistributorSerializer permission_classes = [permissions.DjangoModelPermissions] filter_backends = [filters.SearchFilter] search_fields = ['name'] ## Token Authentication views EXPIRE_HOURS = getattr(settings, 'REST_FRAMEWORK_TOKEN_EXPIRE_HOURS', 24) class ObtainExpiringAuthToken(ObtainAuthToken): throttle_classes = [AnonRateThrottle] def post(self, request): serializer = self.serializer_class(data=request.data) if serializer.is_valid(): try: token = Token.objects.get(user=serializer.validated_data['user']) if token.created < timezone.now() - timedelta(hours=EXPIRE_HOURS): token.delete() except Token.DoesNotExist: pass token, created = Token.objects.get_or_create(user=serializer.validated_data['user']) if not created: # update the created time of the token to keep it valid token.created = datetime.datetime.utcnow() token.save() return Response({'token': token.key, 'username': serializer.validated_data['user'].username, 'expiry': token.created + timedelta(hours=EXPIRE_HOURS)}) return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) class TokenLogout(APIView): def post(self, request, format=None): try: request.user.auth_token.delete() except AttributeError as e: pass return Response(status=status.HTTP_200_OK) def get(self, request, format=None): return Response(status=status.HTTP_405_METHOD_NOT_ALLOWED)