from __future__ import annotations from collections.abc import Mapping import hashlib import json from django.core.exceptions import ValidationError as DjangoValidationError from django.db import transaction from django.utils.dateparse import parse_datetime from rest_framework import generics, serializers, status from rest_framework.response import Response from rest_framework.views import APIView from .models import AuditLog, ChoiceValue, Device, DeviceRegistration, Employee, NumberSequence from .query import SPECIFICATION_RELATIONS from .services import add_audit, update_device_field class SecureAPIResponseMixin: def finalize_response(self, request, response, *args, **kwargs): response = super().finalize_response(request, response, *args, **kwargs) response["Cache-Control"] = "no-store" response["Pragma"] = "no-cache" response["Vary"] = "Authorization" return response class ChoiceSummarySerializer(serializers.ModelSerializer): class Meta: model = ChoiceValue fields = ["id", "label", "color"] class EmployeeSummarySerializer(serializers.ModelSerializer): class Meta: model = Employee fields = ["id", "code", "name"] class DeviceSerializer(serializers.ModelSerializer): camera = serializers.CharField(read_only=True) sound = serializers.CharField(read_only=True) hdmi = serializers.CharField(read_only=True) vga = serializers.CharField(read_only=True) dvi = serializers.CharField(read_only=True) displayport = serializers.CharField(read_only=True) aspect_ratio = serializers.CharField(read_only=True) battery = serializers.CharField(read_only=True) wifi = serializers.CharField(read_only=True) lan = serializers.CharField(read_only=True) display_inches = serializers.DecimalField(max_digits=4, decimal_places=1, read_only=True, allow_null=True) ram_gb = serializers.IntegerField(read_only=True, allow_null=True) ssd_gb = serializers.IntegerField(read_only=True, allow_null=True) hdd_gb = serializers.IntegerField(read_only=True, allow_null=True) cpu = serializers.CharField(read_only=True) hwid = serializers.CharField(read_only=True) ports = serializers.CharField(read_only=True) resolution = serializers.CharField(read_only=True) print_technology = serializers.CharField(read_only=True) duplex = serializers.CharField(read_only=True) color_printing = serializers.CharField(read_only=True) status = ChoiceSummarySerializer(read_only=True) recipient = ChoiceSummarySerializer(read_only=True) location = ChoiceSummarySerializer(read_only=True) device_type = ChoiceSummarySerializer(read_only=True) employee = EmployeeSummarySerializer(read_only=True) class Meta: model = Device fields = [ "number", "stage", "shelf", "status", "date", "recipient", "location", "ticket", "device_type", "price", "battery", "wifi", "lan", "display_inches", "ram_gb", "ssd_gb", "hdd_gb", "cpu", "hwid", "ports", "resolution", "print_technology", "duplex", "color_printing", "aspect_ratio", "camera", "sound", "hdmi", "vga", "dvi", "displayport", "model", "employee", "donor", "comment", "archived_at", "created_at", "updated_at", "revision", ] class StrictSerializer(serializers.Serializer): def to_internal_value(self, data): if isinstance(data, Mapping): unknown = set(data) - set(self.fields) if unknown: names = ", ".join(sorted(unknown)) raise serializers.ValidationError({"unbekannte_felder": f"Nicht erlaubt: {names}"}) return super().to_internal_value(data) def active_choice_field(kind, *, allow_null=True): return serializers.PrimaryKeyRelatedField( queryset=ChoiceValue.objects.filter(kind=kind, active=True), required=False, allow_null=allow_null, ) class DeviceChangesSerializer(StrictSerializer): camera = serializers.ChoiceField(choices=Device.TriState.choices, required=False) sound = serializers.ChoiceField(choices=Device.TriState.choices, required=False) hdmi = serializers.ChoiceField(choices=Device.TriState.choices, required=False) vga = serializers.ChoiceField(choices=Device.TriState.choices, required=False) dvi = serializers.ChoiceField(choices=Device.TriState.choices, required=False) displayport = serializers.ChoiceField(choices=Device.TriState.choices, required=False) aspect_ratio = serializers.CharField(max_length=20, required=False, allow_blank=True) shelf = serializers.CharField(max_length=80, required=False, allow_blank=True) status = active_choice_field(ChoiceValue.Kind.STATUS, allow_null=False) date = serializers.DateField(required=False) recipient = active_choice_field(ChoiceValue.Kind.RECIPIENT) location = active_choice_field(ChoiceValue.Kind.LOCATION) ticket = serializers.CharField(max_length=120, required=False, allow_blank=True) device_type = active_choice_field(ChoiceValue.Kind.DEVICE_TYPE) price = serializers.DecimalField(max_digits=10, decimal_places=2, required=False, allow_null=True, min_value=0) battery = serializers.ChoiceField(choices=Device.TriState.choices, required=False) wifi = serializers.ChoiceField(choices=Device.TriState.choices, required=False) lan = serializers.ChoiceField(choices=Device.TriState.choices, required=False) display_inches = serializers.DecimalField(max_digits=4, decimal_places=1, required=False, allow_null=True, min_value=0) ram_gb = serializers.IntegerField(required=False, allow_null=True, min_value=0) ssd_gb = serializers.IntegerField(required=False, allow_null=True, min_value=0) hdd_gb = serializers.IntegerField(required=False, allow_null=True, min_value=0) cpu = serializers.CharField(max_length=200, required=False, allow_blank=True) hwid = serializers.CharField(max_length=255, required=False, allow_blank=True) ports = serializers.CharField(max_length=255, required=False, allow_blank=True) resolution = serializers.CharField(max_length=80, required=False, allow_blank=True) print_technology = serializers.CharField(max_length=80, required=False, allow_blank=True) duplex = serializers.ChoiceField(choices=Device.TriState.choices, required=False) color_printing = serializers.ChoiceField(choices=Device.TriState.choices, required=False) model = serializers.CharField(max_length=200, required=False, allow_blank=True) employee = serializers.PrimaryKeyRelatedField( queryset=Employee.objects.filter(active=True), required=False, allow_null=True, ) donor = serializers.CharField(max_length=200, required=False, allow_blank=True) comment = serializers.CharField(required=False, allow_blank=True) def validate(self, attrs): if not attrs: raise serializers.ValidationError("Mindestens ein zu änderndes Feld ist erforderlich.") return attrs class DeviceCreateSerializer(DeviceChangesSerializer): request_id = serializers.UUIDField() status = serializers.PrimaryKeyRelatedField( queryset=ChoiceValue.objects.filter(kind=ChoiceValue.Kind.STATUS, active=True), ) device_type = serializers.PrimaryKeyRelatedField( queryset=ChoiceValue.objects.filter(kind=ChoiceValue.Kind.DEVICE_TYPE, active=True), ) class DevicePatchSerializer(DeviceChangesSerializer): expected_updated_at = serializers.DateTimeField(required=False, write_only=True) expected_revision = serializers.IntegerField(required=False, min_value=1, write_only=True) def validate(self, attrs): if not {"expected_updated_at", "expected_revision"} & set(attrs): raise serializers.ValidationError("expected_revision oder expected_updated_at ist erforderlich.") if not set(attrs) - {"expected_updated_at", "expected_revision"}: raise serializers.ValidationError("Mindestens ein zu änderndes Feld ist erforderlich.") return attrs class BulkUpdateSerializer(StrictSerializer): device_numbers = serializers.ListField( child=serializers.IntegerField(min_value=1), min_length=1, max_length=500, ) changes = DeviceChangesSerializer() def validate_device_numbers(self, numbers): if len(numbers) != len(set(numbers)): raise serializers.ValidationError("Gerätenummern dürfen nicht doppelt vorkommen.") return numbers def raw_service_value(value): if isinstance(value, (ChoiceValue, Employee)): return value.pk return value def apply_changes(device, changes, *, user, api_client, expected_updated_at=None, expected_revision=None): current = device for index, (field, value) in enumerate(changes.items()): current = update_device_field( current, field, raw_service_value(value), user, expected_updated_at if index == 0 and expected_updated_at else None, api_client=api_client, expected_revision=expected_revision if index == 0 else None, ) return current class DeviceListView(SecureAPIResponseMixin, generics.ListAPIView): serializer_class = DeviceSerializer @transaction.atomic def post(self, request): serializer = DeviceCreateSerializer(data=request.data) serializer.is_valid(raise_exception=True) values = dict(serializer.validated_data) request_id = values.pop("request_id") digest = hashlib.sha256(json.dumps(request.data, sort_keys=True, separators=(",", ":")).encode()).hexdigest() # Share the existing allocation lock, including across simultaneous retries. NumberSequence.objects.select_for_update().get_or_create(name="device") previous = DeviceRegistration.objects.select_related("device").filter(request_id=request_id).first() if previous: if previous.payload_hash != digest: return Response({"detail": "request_id wurde bereits für andere Daten verwendet."}, status=409) return Response(DeviceSerializer(previous.device).data, status=200) try: device = Device.objects.create(**values) except DjangoValidationError as exc: raise serializers.ValidationError(exc.messages) DeviceRegistration.objects.create(request_id=request_id, payload_hash=digest, device=device) add_audit(request.user, device, AuditLog.Action.CREATE, new=f"Gerät #{device.number}", api_client=request.auth.name) return Response(DeviceSerializer(device).data, status=201) def get_queryset(self): queryset = Device.objects.select_related( "status", "recipient", "location", "device_type", "employee", *SPECIFICATION_RELATIONS, ).order_by("number") params = self.request.query_params if params.get("archived", "false").lower() not in {"1", "true", "yes"}: queryset = queryset.filter(archived_at__isnull=True) if number := params.get("number"): if number.isascii() and number.isdigit() and len(number) <= 10: queryset = queryset.filter(number=int(number)) else: return queryset.none() if prefix := params.get("number_prefix"): if prefix.isascii() and prefix.isdigit() and len(prefix) <= 10: queryset = queryset.filter(number__startswith=prefix) else: return queryset.none() for parameter, field in (("status", "status_id"), ("type", "device_type_id"), ("employee", "employee_id")): value = params.get(parameter) if value: if not value.isascii() or not value.isdigit() or len(value) > 19 or int(value) > 9223372036854775807: return queryset.none() queryset = queryset.filter(**{field: int(value)}) if changed_since := params.get("changed_since"): try: parsed = parse_datetime(changed_since) except ValueError: parsed = None if parsed: queryset = queryset.filter(updated_at__gte=parsed) else: return queryset.none() return queryset class DeviceDetailView(SecureAPIResponseMixin, APIView): def get_object(self, number): return generics.get_object_or_404( Device.objects.select_related("status", "recipient", "location", "device_type", "employee", *SPECIFICATION_RELATIONS), number=number, ) def get(self, request, number): return Response(DeviceSerializer(self.get_object(number)).data) @transaction.atomic def patch(self, request, number): serializer = DevicePatchSerializer(data=request.data) serializer.is_valid(raise_exception=True) values = dict(serializer.validated_data) expected = values.pop("expected_updated_at", None) revision = values.pop("expected_revision", None) device = self.get_object(number) try: device = apply_changes( device, values, user=request.user, api_client=request.auth.name, expected_updated_at=expected, expected_revision=revision, ) except RuntimeError as exc: if str(exc) == "conflict": return Response( {"detail": "Der Datensatz wurde zwischenzeitlich geändert. Bitte neu laden."}, status=status.HTTP_409_CONFLICT, ) raise except DjangoValidationError as exc: raise serializers.ValidationError(exc.messages) device.refresh_from_db() return Response(DeviceSerializer(device).data) class BulkUpdateView(SecureAPIResponseMixin, APIView): @transaction.atomic def post(self, request): serializer = BulkUpdateSerializer(data=request.data) serializer.is_valid(raise_exception=True) numbers = serializer.validated_data["device_numbers"] changes = serializer.validated_data["changes"] devices = list(Device.objects.select_for_update().filter(number__in=numbers, archived_at__isnull=True)) found = {device.number for device in devices} missing = sorted(set(numbers) - found) if missing: raise serializers.ValidationError({"device_numbers": f"Nicht gefunden oder archiviert: {missing}"}) try: for device in devices: apply_changes(device, changes, user=request.user, api_client=request.auth.name) except DjangoValidationError as exc: raise serializers.ValidationError(exc.messages) return Response({"updated": len(devices), "device_numbers": sorted(found)}) class ChoicesView(SecureAPIResponseMixin, APIView): def get(self, request): result = {} for kind, label in ( (ChoiceValue.Kind.STATUS, "statuses"), (ChoiceValue.Kind.RECIPIENT, "recipients"), (ChoiceValue.Kind.LOCATION, "locations"), (ChoiceValue.Kind.DEVICE_TYPE, "device_types"), ): values = ChoiceValue.objects.filter(kind=kind, active=True).order_by("sort_order", "label") result[label] = ChoiceSummarySerializer(values, many=True).data employees = Employee.objects.filter(active=True).order_by("code") result["employees"] = EmployeeSummarySerializer(employees, many=True).data result["tri_state_values"] = [{"value": value, "label": label} for value, label in Device.TriState.choices] return Response(result)