-
Notifications
You must be signed in to change notification settings - Fork 27
/
views.py
75 lines (64 loc) · 2.93 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
70
71
72
73
74
75
from dateutil.relativedelta import relativedelta
from django.utils import timezone
from drf_yasg import openapi
from drf_yasg.utils import swagger_auto_schema
from pydash import compact
from rest_framework import status
from rest_framework.permissions import IsAdminUser
from rest_framework.response import Response
from rest_framework.views import APIView
from core.common.tasks import resources_report
from core.common.views import BaseAPIView
from core.tasks.models import Task
from core.tasks.serializers import TaskBriefSerializer
from core.users.reports import UserReport
class ResourcesReportJobView(APIView): # pragma: no cover
permission_classes = (IsAdminUser, )
@staticmethod
@swagger_auto_schema(
operation_description='Mails CSV of OCL resources usage for given period of time',
operation_summary='Reports Resources Usage on the env for given period of time',
request_body=openapi.Schema(
type=openapi.TYPE_OBJECT,
properties={
'start_date': openapi.Schema(
type=openapi.TYPE_STRING,
description='YYYY-MM-DD (default: 1st of last month)',
default=(timezone.now().replace(day=1) - relativedelta(months=1)).strftime('%Y-%m-%d')
),
'end_date': openapi.Schema(
type=openapi.TYPE_STRING,
description='YYYY-MM-DD (default: 1st of current month)',
default=timezone.now().replace(day=1).strftime('%Y-%m-%d')
),
}
)
)
def post(request):
task = Task.make_new(queue='default', user=request.user, name=resources_report.__name__)
resources_report.apply_async(
(request.data.get('start_date'), request.data.get('end_date')), queue=task.queue, task_id=task.id)
task.refresh_from_db()
return Response(TaskBriefSerializer(task).data, status=status.HTTP_202_ACCEPTED)
class AuthoredView(BaseAPIView): # pragma: no cover
permission_classes = (IsAdminUser, )
@staticmethod
@swagger_auto_schema(
operation_description='Returns count of content created/updated by the user(s)',
operation_summary='Resources Authored Summary for user(s)',
request_body=openapi.Schema(
type=openapi.TYPE_OBJECT,
properties={
'usernames': openapi.Schema(type=openapi.TYPE_STRING, description='usernames comma separated')
}
)
)
def post(request):
if not request.data:
return Response(status=status.HTTP_400_BAD_REQUEST)
usernames = request.data.get('usernames', None)
if not usernames:
return Response(status=status.HTTP_400_BAD_REQUEST)
if isinstance(usernames, str):
usernames = compact([username.strip() for username in usernames.split(',')])
return Response(UserReport.get_authoring_report(usernames), status.HTTP_200_OK)