1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91
|
from rest_framework_extensions.cache.mixins import CacheResponseMixin
# from rest_framework_extensions.etag.mixins import ReadOnlyETAGMixin, ETAGMixin
from rest_framework_extensions.bulk_operations.mixins import ListUpdateModelMixin, ListDestroyModelMixin
from rest_framework_extensions.settings import extensions_api_settings
from django.core.exceptions import ValidationError
from django.http import Http404
import uuid
class DetailSerializerMixin:
"""
Add custom serializer for detail view
"""
serializer_detail_class = None
queryset_detail = None
def get_serializer_class(self):
error_message = "'{0}' should include a 'serializer_detail_class' attribute".format(
self.__class__.__name__)
assert self.serializer_detail_class is not None, error_message
if self._is_request_to_detail_endpoint():
return self.serializer_detail_class
else:
return super().get_serializer_class()
def get_queryset(self, *args, **kwargs):
if self._is_request_to_detail_endpoint() and self.queryset_detail is not None:
return self.queryset_detail.all() # todo: test all()
else:
return super().get_queryset(*args, **kwargs)
def _is_request_to_detail_endpoint(self):
if hasattr(self, 'lookup_url_kwarg'):
lookup = self.lookup_url_kwarg or self.lookup_field
return lookup and lookup in self.kwargs
class PaginateByMaxMixin:
def get_page_size(self, request):
if self.page_size_query_param and self.max_page_size and request.query_params.get(self.page_size_query_param) == 'max':
return self.max_page_size
return super().get_page_size(request)
# class ReadOnlyCacheResponseAndETAGMixin(ReadOnlyETAGMixin, CacheResponseMixin):
# pass
# class CacheResponseAndETAGMixin(ETAGMixin, CacheResponseMixin):
# pass
class NestedViewSetMixin:
def get_queryset(self):
return self.filter_queryset_by_parents_lookups(
super().get_queryset()
)
def filter_queryset_by_parents_lookups(self, queryset):
parents_query_dict = self.get_parents_query_dict()
if parents_query_dict:
try:
# Try to validate UUID fields before filtering
cleaned_dict = {}
for key, value in parents_query_dict.items():
if 'uuid' in key.lower() or key.endswith('_code'):
try:
# Try to validate as UUID
cleaned_dict[key] = uuid.UUID(str(value))
except ValueError:
raise Http404
else:
cleaned_dict[key] = value
return queryset.filter(**cleaned_dict)
except (ValueError, ValidationError):
raise Http404
else:
return queryset
def get_parents_query_dict(self):
result = {}
for kwarg_name, kwarg_value in self.kwargs.items():
if kwarg_name.startswith(extensions_api_settings.DEFAULT_PARENT_LOOKUP_KWARG_NAME_PREFIX):
query_lookup = kwarg_name.replace(
extensions_api_settings.DEFAULT_PARENT_LOOKUP_KWARG_NAME_PREFIX,
'',
1
)
query_value = kwarg_value
result[query_lookup] = query_value
return result
|