-
Notifications
You must be signed in to change notification settings - Fork 27
/
views.py
69 lines (53 loc) · 2.54 KB
/
views.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
from drf_yasg.utils import swagger_auto_schema
from pydash import compact
from rest_framework.generics import RetrieveAPIView, DestroyAPIView
from rest_framework.permissions import IsAuthenticated, IsAdminUser
from core.common.exceptions import Http400
from core.common.mixins import ListWithHeadersMixin
from core.common.swagger_parameters import page_param, verbose_param, task_state_param, limit_param, \
task_start_date_param, q_param
from core.common.utils import from_string_to_date
from core.common.views import BaseAPIView
from core.tasks.models import Task
from core.tasks.serializers import TaskDetailSerializer, TaskListSerializer
class AbstractTaskView(BaseAPIView, RetrieveAPIView):
queryset = Task.objects.filter()
is_searchable = False
permission_classes = (IsAuthenticated,)
default_qs_sort_attr = '-created_at'
def get_serializer_class(self):
return TaskDetailSerializer if self.is_verbose() else TaskListSerializer
def get_queryset(self):
states = compact((self.request.query_params.get('state', None) or '').split(','))
start_date = from_string_to_date(self.request.query_params.get('start_date', None))
search_str = self.request.query_params.get('q', None)
queryset = self.queryset.filter(started_at__gte=start_date) if start_date else self.queryset
if states:
queryset = queryset.filter(state__in=states)
if search_str:
queryset = queryset.filter(name__icontains=search_str)
return queryset.order_by('-created_at')
class AbstractTaskListView(AbstractTaskView, ListWithHeadersMixin):
@swagger_auto_schema(
manual_parameters=[q_param, task_state_param, task_start_date_param, page_param, limit_param, verbose_param]
)
def get(self, request, *args, **kwargs):
return self.list(request, *args, **kwargs)
class TaskView(AbstractTaskView, DestroyAPIView):
lookup_field = 'id'
lookup_url_kwarg = 'task_id'
pk_field = 'id'
def perform_destroy(self, instance):
if not instance.has_access(self.request.user):
self.permission_denied(self.request)
if instance.is_finished():
raise Http400('Task is already finished.')
try:
instance.revoke()
except Exception as ex:
raise Http400({'errors': ex.args}) from ex
class UserTaskListView(AbstractTaskListView):
def get_queryset(self):
return super().get_queryset().filter(created_by=self.request.user)
class TaskListView(AbstractTaskListView):
permission_classes = (IsAdminUser,)