forked from chibisov/drf-extensions
-
Notifications
You must be signed in to change notification settings - Fork 0
/
mixins.py
85 lines (70 loc) · 3.21 KB
/
mixins.py
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
# -*- coding: utf-8 -*-
from rest_framework_extensions.cache.mixins import CacheResponseMixin
from rest_framework_extensions.etag.mixins import ReadOnlyETAGMixin, ETAGMixin
from rest_framework_extensions.utils import get_rest_framework_features
from rest_framework_extensions.bulk_operations.mixins import ListDestroyModelMixin, ListUpdateModelMixin
from rest_framework_extensions.settings import extensions_api_settings
from django.http import Http404
class DetailSerializerMixin(object):
"""
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(DetailSerializerMixin, self).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(DetailSerializerMixin, self).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
else: # DRF 2 compatibility
lookup = self.pk_url_kwarg or self.slug_url_kwarg
return lookup and lookup in self.kwargs
class PaginateByMaxMixin(object):
def get_paginate_by(self, *args, **kwargs):
if (get_rest_framework_features()['max_paginate_by'] and
self.paginate_by_param and
self.max_paginate_by and
self.request.QUERY_PARAMS.get(self.paginate_by_param) == 'max'):
return self.max_paginate_by
else:
return super(PaginateByMaxMixin, self).get_paginate_by(*args, **kwargs)
class ReadOnlyCacheResponseAndETAGMixin(ReadOnlyETAGMixin, CacheResponseMixin):
pass
class CacheResponseAndETAGMixin(ETAGMixin, CacheResponseMixin):
pass
class NestedViewSetMixin(object):
def get_queryset(self):
return self.filter_queryset_by_parents_lookups(
super(NestedViewSetMixin, self).get_queryset()
)
def filter_queryset_by_parents_lookups(self, queryset):
parents_query_dict = self.get_parents_query_dict()
if parents_query_dict:
try:
return queryset.filter(**parents_query_dict)
except ValueError:
raise Http404
else:
return queryset
def get_parents_query_dict(self):
result = {}
for kwarg_name in self.kwargs:
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 = self.kwargs.get(kwarg_name)
result[query_lookup] = query_value
return result