116 lines
4.5 KiB
Python
116 lines
4.5 KiB
Python
from django.shortcuts import render
|
|
from django.contrib.auth.models import User, Group
|
|
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
|
|
|
|
# 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]
|
|
|
|
class PartsComponentViewSet(viewsets.ModelViewSet):
|
|
queryset = parts_models.Component.objects.all()
|
|
serializer_class = ComponentSerializer
|
|
permission_classes = [permissions.DjangoModelPermissions]
|
|
|
|
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]
|
|
|
|
class PartsDistributorviewSet(viewsets.ModelViewSet):
|
|
queryset = parts_models.Distributor.objects.all()
|
|
serializer_class = DistributorSerializer
|
|
permission_classes = [permissions.DjangoModelPermissions]
|
|
|
|
|
|
## 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)
|