161 lines
5.6 KiB
Python
Executable File
161 lines
5.6 KiB
Python
Executable File
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
|