from rest_framework import generics, status from rest_framework.response import Response from django_filters.rest_framework import DjangoFilterBackend from .models import Device from .serializers import ( CreateDeviceSerializer, DeviceSerializer, ReadOnlyDeviceSerializer, BlockDeviceSerializer, ) from api.mixins import StaffEditorPermissionMixin from .filters import DeviceFilter import re class DeviceListCreateAPIView( StaffEditorPermissionMixin, generics.ListCreateAPIView, ): queryset = Device.objects.select_related("user").prefetch_related("payments").all() serializer_class = CreateDeviceSerializer filter_backends = [DjangoFilterBackend] filterset_fields = "__all__" filterset_class = DeviceFilter def list(self, request, *args, **kwargs): queryset = self.filter_queryset(self.get_queryset()) # Filter devices by the logged-in user unless the user is a superuser if not request.user.is_superuser: queryset = queryset.filter(user=request.user) page = self.paginate_queryset(queryset) if page is not None: serializer = self.get_serializer(page, many=True) return self.get_paginated_response(serializer.data) serializer = self.get_serializer(queryset, many=True) return Response(serializer.data) def get_serializer_class(self) -> type: if self.request.method == "POST": return CreateDeviceSerializer return DeviceSerializer # @method_decorator(cache_page(10)) def create(self, request, *args, **kwargs): mac = request.data.get("mac", None) if not re.match(r"^([0-9A-Fa-f]{2}([.:-]?)){5}[0-9A-Fa-f]{2}$", mac): return Response({"message": "Invalid mac address."}, status=400) if Device.objects.filter(mac=mac).exists(): return Response( {"message": "Device with this mac address already exists."}, status=400 ) return super().create(request, *args, **kwargs) def perform_create(self, serializer): serializer.save(user=self.request.user) class DeviceDetailAPIView(StaffEditorPermissionMixin, generics.RetrieveAPIView): queryset = Device.objects.select_related("user").all() serializer_class = ReadOnlyDeviceSerializer lookup_field = "pk" class DeviceUpdateAPIView(StaffEditorPermissionMixin, generics.UpdateAPIView): queryset = Device.objects.all() serializer_class = CreateDeviceSerializer lookup_field = "pk" def update(self, request, **kwargs): instance = self.get_object() user_id = request.user.id if not request.user.is_superuser and instance.user_id != user_id: return Response( {"message": "You are not authorized to update this device."}, status=403, ) # Validate MAC address format mac = request.data.get("mac") if mac and not re.match(r"^([0-9A-Fa-f]{2}([.:-]?)){5}[0-9A-Fa-f]{2}$", mac): return Response({"message": "Invalid MAC address"}, status=400) serializer = self.get_serializer(instance, data=request.data, partial=True) serializer.is_valid(raise_exception=True) self.perform_update(serializer) return Response(serializer.data) class DeviceBlockAPIView(StaffEditorPermissionMixin, generics.UpdateAPIView): queryset = Device.objects.all() serializer_class = BlockDeviceSerializer lookup_field = "pk" def update(self, request, *args, **kwargs): # Pass 'partial=True' to allow partial updates user_id = request.user.id instance = self.get_object() if not request.user.is_superuser and instance.user_id != user_id: return Response( {"message": "You are not authorized to block this device."}, status=403, ) blocked = request.data.get("blocked", None) if blocked is None: return Response({"message": "Blocked field is required."}, status=400) if not isinstance(blocked, bool): return Response({"message": "Blocked field must be a boolean."}, status=400) instance.blocked = blocked instance.save() serializer = self.get_serializer(instance, data=request.data, partial=False) serializer.is_valid(raise_exception=True) self.perform_update(serializer) return Response(serializer.data) class DeviceDestroyAPIView(StaffEditorPermissionMixin, generics.DestroyAPIView): queryset = Device.objects.all() serializer_class = DeviceSerializer lookup_field = "pk" def destroy(self, request, *args, **kwargs): instance = self.get_object() device_name = instance.name self.perform_destroy(instance) return Response( {"message": f"Device '{device_name}' deleted."}, status=status.HTTP_200_OK, )