283 lines
13 KiB
Python
283 lines
13 KiB
Python
from __future__ import annotations
|
|
|
|
from collections.abc import Mapping
|
|
|
|
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 ChoiceValue, Device, Employee
|
|
from .query import SPECIFICATION_RELATIONS
|
|
from .services import 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):
|
|
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", "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):
|
|
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 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
|
|
|
|
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)
|