Initial commit from ZIP

evg-outbo-api/outbo/apps/core/viewsets.py

@@ -0,0 +1,160 @@
+import six
+from django.db import transaction
+from django.db.models import Model
+from rest_framework import status, mixins
+from rest_framework.response import Response
+from rest_framework.settings import api_settings
+from rest_framework.viewsets import GenericViewSet
+from rest_framework_extensions.mixins import (
+    NestedViewSetMixin as BaseNestedViewSetMixin,
+)
+from rest_framework_extensions.settings import extensions_api_settings
+from rules.contrib.rest_framework import (
+    AutoPermissionViewSetMixin as BaseAutoPermissionViewSetMixin,
+)
+
+
+class AutoPermissionViewSetMixin(BaseAutoPermissionViewSetMixin):
+    permission_type_map = {
+        **BaseAutoPermissionViewSetMixin.permission_type_map,
+        "metadata": None,
+    }
+
+
+class CreateModelMixin:
+    def create(self, request, *args, **kwargs):
+        if isinstance(request.data, list):
+            with transaction.atomic():
+                response = []
+                for item in request.data:
+                    write_serializer = self.get_serializer(data=item)
+                    write_serializer.is_valid(raise_exception=True)
+                    instance = self.perform_create(serializer=write_serializer)
+                    read_serializer = self.get_read_serializer(instance=instance)
+                    response.append(read_serializer.data)
+                return Response(response, status=status.HTTP_201_CREATED)
+
+        else:
+            write_serializer = self.get_serializer(data=request.data)
+            write_serializer.is_valid(raise_exception=True)
+
+            instance = self.perform_create(serializer=write_serializer)
+
+            read_serializer = self.get_read_serializer(instance=instance)
+            headers = self.get_success_headers(read_serializer.data)
+
+            return Response(
+                read_serializer.data, status=status.HTTP_201_CREATED, headers=headers
+            )
+
+    def perform_create(self, *, serializer):
+        return serializer.save()
+
+    def get_success_headers(self, data):
+        try:
+            return {"Location": str(data[api_settings.URL_FIELD_NAME])}
+        except (TypeError, KeyError):
+            return {}
+
+
+class UpdateModelMixin:
+    def update(self, request, *args, **kwargs):
+        instance = self.get_object()
+        write_serializer = self.get_serializer(
+            instance, data=request.data, partial=False
+        )
+        write_serializer.is_valid(raise_exception=True)
+
+        instance = write_serializer.save()
+
+        if getattr(instance, "_prefetched_objects_cache", None):
+            # If 'prefetch_related' has been applied to a queryset, we need to
+            # forcibly invalidate the prefetch cache on the instance.
+            instance._prefetched_objects_cache = {}
+
+        read_serializer = self.get_read_serializer(instance=instance)
+
+        return Response(read_serializer.data)
+
+
+class PartialUpdateModelMixin:
+    def partial_update(self, request, *args, **kwargs):
+        instance = self.get_object()
+        write_serializer = self.get_serializer(
+            instance, data=request.data, partial=True
+        )
+        write_serializer.is_valid(raise_exception=True)
+
+        instance = write_serializer.save()
+
+        if getattr(instance, "_prefetched_objects_cache", None):
+            # If 'prefetch_related' has been applied to a queryset, we need to
+            # forcibly invalidate the prefetch cache on the instance.
+            instance._prefetched_objects_cache = {}
+
+        read_serializer = self.get_read_serializer(instance=instance)
+
+        return Response(read_serializer.data)
+
+
+class RetrieveModelMixin(mixins.RetrieveModelMixin):
+    pass
+
+
+class ListModelMixin(mixins.ListModelMixin):
+    pass
+
+
+class DestroyModelMixin(mixins.DestroyModelMixin):
+    pass
+
+
+class MultipleSerializerViewSet(GenericViewSet):
+    serializer_class_set = None
+    serializer_map = {
+        "list": "List",
+        "retrieve": "Retrieve",
+        "create": "Create",
+        "partial_update": "Update",
+    }
+
+    def get_serializer_class(self):
+        assert (
+            self.serializer_class_set is not None
+        ), "serializer_class_set should be defined"
+
+        if self.action not in self.serializer_map:
+            raise KeyError(
+                f"""serializer_map does not contain serializer for "{self.action}" method"""
+            )
+        return getattr(self.serializer_class_set, self.serializer_map[self.action])
+
+    def get_read_serializer(self, *args, **kwargs):
+        assert (
+            self.serializer_class_set is not None
+        ), "serializer_class_set should be defined"
+        if self.action != "create":
+            if getattr(self.serializer_class_set, "Retrieve", None) is None:
+                raise KeyError(
+                    f"""serializer_class_set does not contain "Retrieve" serializer"""
+                )
+            serializer_class = self.serializer_class_set.Retrieve
+        else:
+            if (
+                getattr(self.serializer_class_set, "Retrieve", None) is None
+                and getattr(self.serializer_class_set, "AfterCreate", None) is None
+            ):
+                raise KeyError(
+                    f"""serializer_class_set does not contain "Retrieve" nor "AfterCreate" serializer and is used in "create" action."""
+                )
+            if getattr(self.serializer_class_set, "AfterCreate", None) is not None:
+                serializer_class = self.serializer_class_set.AfterCreate
+            else:
+                serializer_class = self.serializer_class_set.Retrieve
+
+        kwargs["context"] = self.get_serializer_context()
+        return serializer_class(*args, **kwargs)
+
+
+class NestedViewSetMixin(BaseNestedViewSetMixin):
+    pass