11896297726

Committed 18 Nov 2024 04:03PM UTC coverage: 93.756% (-0.05%) from 93.806%

Build # 11896297726

Build Type

Pull #266

github

Committed by

web-flow

Commit Message

Merge c6333a03c into 509b1685c

Pull Request Pull Request #266: update default file setting

Run Details

3799 of 4052 relevant lines covered (93.76%)

0.94 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

93.69

/data_aggregator/views/api/analytics.py

# Copyright 2024 UW-IT, University of Washington
# SPDX-License-Identifier: Apache-2.0


from django.conf import settings
from django.db.models import F
from django.utils.decorators import method_decorator
from rest_framework.generics import GenericAPIView
from rest_framework.pagination import PageNumberPagination
from rest_framework.renderers import JSONRenderer
from uw_saml.decorators import group_required
from data_aggregator.models import Assignment, Participation, User
from data_aggregator.serializers import ParticipationSerializer, \
    AssignmentSerializer, UserSerializer

"""
Analytics API
"""


class AnalyticsResultsSetPagination(PageNumberPagination):
    page_size = 1000
    page_size_query_param = 'page_size'
    max_page_size = 2500


@method_decorator(group_required(settings.DATA_AGGREGATOR_ACCESS_GROUP),
                  name='dispatch')
class BaseAnalyticsAPIView(GenericAPIView):

    renderer_classes = [JSONRenderer]
    pagination_class = AnalyticsResultsSetPagination

    def get_assignment_queryset(self):
        queryset = (
            Assignment.objects.select_related()
            .annotate(sis_account_id=F('course__sis_account_id'))
            .annotate(sis_term_id=F('week__term__sis_term_id'))
        )
        return queryset

    def get_participation_queryset(self):
        queryset = (
            Participation.objects.select_related()
            .annotate(sis_account_id=F('course__sis_account_id'))
            .annotate(sis_term_id=F('week__term__sis_term_id'))
        )
        return queryset


class AccountParticipationView(BaseAnalyticsAPIView):
    '''
    API endpoint returning participation analytics for a account

    /api/[version]/account/[sis-account-id]/participation/

    Endpoint accepts the following query parameters:
    * sis_term_id: limit results to a term
    * week: limit results to a week in term
    '''
    def get(self, request, version, sis_account_id):
        queryset = (
            self.get_participation_queryset()
            .filter(sis_account_id__startswith=sis_account_id))
        sis_term_id = request.GET.get("sis_term_id")
        if sis_term_id:
            queryset = queryset.filter(sis_term_id=sis_term_id)
        week = request.GET.get("week")
        if week:
            queryset = queryset.filter(week__week=week)
        paginated_queryset = self.paginate_queryset(queryset)
        serializer = ParticipationSerializer(paginated_queryset, many=True)
        return self.get_paginated_response(serializer.data)


class AccountAssignmentView(BaseAnalyticsAPIView):
    '''
    API endpoint returning assignment analytics for a account

    /api/[version]/account/[sis-account-id]/assignment/

    Endpoint accepts the following query parameters:
    * sis_term_id: limit results to a term
    * week: limit results to a week in term
    '''
    def get(self, request, version, sis_account_id):
        queryset = (
            self.get_assignment_queryset()
            .filter(sis_account_id__startswith=sis_account_id))
        sis_term_id = request.GET.get("sis_term_id")
        if sis_term_id:
            queryset = queryset.filter(sis_term_id=sis_term_id)
        week = request.GET.get("week")
        if week:
            queryset = queryset.filter(week__week=week)

        paginated_queryset = self.paginate_queryset(queryset)
        serializer = AssignmentSerializer(paginated_queryset, many=True)
        return self.get_paginated_response(serializer.data)


class TermParticipationView(BaseAnalyticsAPIView):
    '''
    API endpoint returning participation analytics for a term

    /api/[version]/term/[sis-term-id]/participation/

    Endpoint accepts the following query parameters:
    * week: limit results to a week in term
    '''
    def get(self, request, version, sis_term_id):
        queryset = (
            self.get_participation_queryset()
            .filter(sis_term_id=sis_term_id))
        week = request.GET.get("week")
        if week:
            queryset = queryset.filter(week__week=week)

        paginated_queryset = self.paginate_queryset(queryset)
        serializer = ParticipationSerializer(paginated_queryset, many=True)
        return self.get_paginated_response(serializer.data)


class TermAssignmentView(BaseAnalyticsAPIView):
    '''
    API endpoint returning assignment analytics for a term

    /api/[version]/term/[sis-term-id]/assignment/

    Endpoint accepts the following query parameters:
    * week: limit results to a week in term
    '''
    def get(self, request, version, sis_term_id):
        queryset = (
            self.get_assignment_queryset()
            .filter(sis_term_id=sis_term_id))
        week = request.GET.get("week")
        if week:
            queryset = queryset.filter(week__week=week)

        paginated_queryset = self.paginate_queryset(queryset)
        serializer = AssignmentSerializer(paginated_queryset, many=True)
        return self.get_paginated_response(serializer.data)


class UserView(GenericAPIView):

    renderer_classes = [JSONRenderer]
    pagination_class = AnalyticsResultsSetPagination

    def get(self, request, version):
        queryset = User.objects.select_related()
        canvas_user_id = request.GET.get("canvas_user_id")
        if (canvas_user_id):
            queryset = queryset.filter(canvas_user_id=canvas_user_id)
        has_analytics = request.GET.get("has_analytics")
        if (has_analytics is not None):
            assign_analytic_users = Assignment.objects.values("user")
            partic_analytic_users = Participation.objects.values("user")
            if has_analytics.lower() == "true":
                queryset = queryset.filter(
                    id__in=assign_analytic_users)
                queryset = queryset.filter(
                    id__in=partic_analytic_users)
            elif has_analytics.lower() == "false":
                queryset = queryset.exclude(
                    id__in=assign_analytic_users)
                queryset = queryset.exclude(
                    id__in=partic_analytic_users)
        paginated_queryset = self.paginate_queryset(queryset)
        serializer = UserSerializer(paginated_queryset, many=True)
        return self.get_paginated_response(serializer.data)


class UserParticipationView(BaseAnalyticsAPIView):
    '''
    API endpoint returning participation analytics for a particular user

    /api/[version]/user/[sis-user-id]/participation/

    Endpoint accepts the following query parameters:
    * sis_term_id: limit results to a term
    * week: limit results to a week in term
    '''
    def get(self, request, version, sis_user_id):
        queryset = (
            self.get_participation_queryset()
            .filter(user__sis_user_id=sis_user_id))
        sis_term_id = request.GET.get("sis_term_id")
        if sis_term_id:
            queryset = queryset.filter(sis_term_id=sis_term_id)
        week = request.GET.get("week")
        if week:
            queryset = queryset.filter(week__week=week)

        paginated_queryset = self.paginate_queryset(queryset)
        serializer = ParticipationSerializer(paginated_queryset, many=True)
        return self.get_paginated_response(serializer.data)


class UserAssignmentView(BaseAnalyticsAPIView):
    '''
    API endpoint returning assignment analytics for a particular user

    /api/[version]/user/[sis-user-id]/assignment/

    Endpoint accepts the following query parameters:
    * sis_term_id: limit results to a term
    * week: limit results to a week in term
    '''
    def get(self, request, version, sis_user_id):
        queryset = (
            self.get_assignment_queryset()
            .filter(user__sis_user_id=sis_user_id))
        sis_term_id = request.GET.get("sis_term_id")
        if sis_term_id:
            queryset = queryset.filter(sis_term_id=sis_term_id)
        week = request.GET.get("week")
        if week:
            queryset = queryset.filter(week__week=week)

        paginated_queryset = self.paginate_queryset(queryset)
        serializer = AssignmentSerializer(paginated_queryset, many=True)
        return self.get_paginated_response(serializer.data)

1	# Copyright 2024 UW-IT, University of Washington
2	# SPDX-License-Identifier: Apache-2.0
3
4
5	from django.conf import settings	1✔
6	from django.db.models import F	1✔
7	from django.utils.decorators import method_decorator	1✔
8	from rest_framework.generics import GenericAPIView	1✔
9	from rest_framework.pagination import PageNumberPagination	1✔
10	from rest_framework.renderers import JSONRenderer	1✔
11	from uw_saml.decorators import group_required	1✔
12	from data_aggregator.models import Assignment, Participation, User	1✔
13	from data_aggregator.serializers import ParticipationSerializer, \	1✔
14	AssignmentSerializer, UserSerializer
15
16	"""	1✔
17	Analytics API
18	"""
19
20
21	class AnalyticsResultsSetPagination(PageNumberPagination):	1✔
22	page_size = 1000	1✔
23	page_size_query_param = 'page_size'	1✔
24	max_page_size = 2500	1✔
25
26
27	@method_decorator(group_required(settings.DATA_AGGREGATOR_ACCESS_GROUP),	1✔
28	name='dispatch')
29	class BaseAnalyticsAPIView(GenericAPIView):	1✔
30
31	renderer_classes = [JSONRenderer]	1✔
32	pagination_class = AnalyticsResultsSetPagination	1✔
33
34	def get_assignment_queryset(self):	1✔
35	queryset = (	1✔
36	Assignment.objects.select_related()
37	.annotate(sis_account_id=F('course__sis_account_id'))
38	.annotate(sis_term_id=F('week__term__sis_term_id'))
39	)
40	return queryset	1✔
41
42	def get_participation_queryset(self):	1✔
43	queryset = (	1✔
44	Participation.objects.select_related()
45	.annotate(sis_account_id=F('course__sis_account_id'))
46	.annotate(sis_term_id=F('week__term__sis_term_id'))
47	)
48	return queryset	1✔
49
50
51	class AccountParticipationView(BaseAnalyticsAPIView):	1✔
52	'''
53	API endpoint returning participation analytics for a account
54
55	/api/[version]/account/[sis-account-id]/participation/
56
57	Endpoint accepts the following query parameters:
58	* sis_term_id: limit results to a term
59	* week: limit results to a week in term
60	'''
61	def get(self, request, version, sis_account_id):	1✔
62	queryset = (	1✔
63	self.get_participation_queryset()
64	.filter(sis_account_id__startswith=sis_account_id))
65	sis_term_id = request.GET.get("sis_term_id")	1✔
66	if sis_term_id:	1✔
67	queryset = queryset.filter(sis_term_id=sis_term_id)	×
68	week = request.GET.get("week")	1✔
69	if week:	1✔
70	queryset = queryset.filter(week__week=week)	×
71	paginated_queryset = self.paginate_queryset(queryset)	1✔
72	serializer = ParticipationSerializer(paginated_queryset, many=True)	1✔
73	return self.get_paginated_response(serializer.data)	1✔
74
75
76	class AccountAssignmentView(BaseAnalyticsAPIView):	1✔
77	'''
78	API endpoint returning assignment analytics for a account
79
80	/api/[version]/account/[sis-account-id]/assignment/
81
82	Endpoint accepts the following query parameters:
83	* sis_term_id: limit results to a term
84	* week: limit results to a week in term
85	'''
86	def get(self, request, version, sis_account_id):	1✔
87	queryset = (	1✔
88	self.get_assignment_queryset()
89	.filter(sis_account_id__startswith=sis_account_id))
90	sis_term_id = request.GET.get("sis_term_id")	1✔
91	if sis_term_id:	1✔
92	queryset = queryset.filter(sis_term_id=sis_term_id)	×
93	week = request.GET.get("week")	1✔
94	if week:	1✔
95	queryset = queryset.filter(week__week=week)	×
96
97	paginated_queryset = self.paginate_queryset(queryset)	1✔
98	serializer = AssignmentSerializer(paginated_queryset, many=True)	1✔
99	return self.get_paginated_response(serializer.data)	1✔
100
101
102	class TermParticipationView(BaseAnalyticsAPIView):	1✔
103	'''
104	API endpoint returning participation analytics for a term
105
106	/api/[version]/term/[sis-term-id]/participation/
107
108	Endpoint accepts the following query parameters:
109	* week: limit results to a week in term
110	'''
111	def get(self, request, version, sis_term_id):	1✔
112	queryset = (	1✔
113	self.get_participation_queryset()
114	.filter(sis_term_id=sis_term_id))
115	week = request.GET.get("week")	1✔
116	if week:	1✔
117	queryset = queryset.filter(week__week=week)	×
118
119	paginated_queryset = self.paginate_queryset(queryset)	1✔
120	serializer = ParticipationSerializer(paginated_queryset, many=True)	1✔
121	return self.get_paginated_response(serializer.data)	1✔
122
123
124	class TermAssignmentView(BaseAnalyticsAPIView):	1✔
125	'''
126	API endpoint returning assignment analytics for a term
127
128	/api/[version]/term/[sis-term-id]/assignment/
129
130	Endpoint accepts the following query parameters:
131	* week: limit results to a week in term
132	'''
133	def get(self, request, version, sis_term_id):	1✔
134	queryset = (	1✔
135	self.get_assignment_queryset()
136	.filter(sis_term_id=sis_term_id))
137	week = request.GET.get("week")	1✔
138	if week:	1✔
139	queryset = queryset.filter(week__week=week)	×
140
141	paginated_queryset = self.paginate_queryset(queryset)	1✔
142	serializer = AssignmentSerializer(paginated_queryset, many=True)	1✔
143	return self.get_paginated_response(serializer.data)	1✔
144
145
146	class UserView(GenericAPIView):	1✔
147
148	renderer_classes = [JSONRenderer]	1✔
149	pagination_class = AnalyticsResultsSetPagination	1✔
150
151	def get(self, request, version):	1✔
152	queryset = User.objects.select_related()	1✔
153	canvas_user_id = request.GET.get("canvas_user_id")	1✔
154	if (canvas_user_id):	1✔
155	queryset = queryset.filter(canvas_user_id=canvas_user_id)	×
156	has_analytics = request.GET.get("has_analytics")	1✔
157	if (has_analytics is not None):	1✔
158	assign_analytic_users = Assignment.objects.values("user")	1✔
159	partic_analytic_users = Participation.objects.values("user")	1✔
160	if has_analytics.lower() == "true":	1✔
161	queryset = queryset.filter(	1✔
162	id__in=assign_analytic_users)
163	queryset = queryset.filter(	1✔
164	id__in=partic_analytic_users)
165	elif has_analytics.lower() == "false":	1✔
166	queryset = queryset.exclude(	1✔
167	id__in=assign_analytic_users)
168	queryset = queryset.exclude(	1✔
169	id__in=partic_analytic_users)
170	paginated_queryset = self.paginate_queryset(queryset)	1✔
171	serializer = UserSerializer(paginated_queryset, many=True)	1✔
172	return self.get_paginated_response(serializer.data)	1✔
173
174
175	class UserParticipationView(BaseAnalyticsAPIView):	1✔
176	'''
177	API endpoint returning participation analytics for a particular user
178
179	/api/[version]/user/[sis-user-id]/participation/
180
181	Endpoint accepts the following query parameters:
182	* sis_term_id: limit results to a term
183	* week: limit results to a week in term
184	'''
185	def get(self, request, version, sis_user_id):	1✔
186	queryset = (	1✔
187	self.get_participation_queryset()
188	.filter(user__sis_user_id=sis_user_id))
189	sis_term_id = request.GET.get("sis_term_id")	1✔
190	if sis_term_id:	1✔
191	queryset = queryset.filter(sis_term_id=sis_term_id)	1✔
192	week = request.GET.get("week")	1✔
193	if week:	1✔
194	queryset = queryset.filter(week__week=week)	1✔
195
196	paginated_queryset = self.paginate_queryset(queryset)	1✔
197	serializer = ParticipationSerializer(paginated_queryset, many=True)	1✔
198	return self.get_paginated_response(serializer.data)	1✔
199
200
201	class UserAssignmentView(BaseAnalyticsAPIView):	1✔
202	'''
203	API endpoint returning assignment analytics for a particular user
204
205	/api/[version]/user/[sis-user-id]/assignment/
206
207	Endpoint accepts the following query parameters:
208	* sis_term_id: limit results to a term
209	* week: limit results to a week in term
210	'''
211	def get(self, request, version, sis_user_id):	1✔
212	queryset = (	1✔
213	self.get_assignment_queryset()
214	.filter(user__sis_user_id=sis_user_id))
215	sis_term_id = request.GET.get("sis_term_id")	1✔
216	if sis_term_id:	1✔
217	queryset = queryset.filter(sis_term_id=sis_term_id)	1✔
218	week = request.GET.get("week")	1✔
219	if week:	1✔
220	queryset = queryset.filter(week__week=week)	1✔
221
222	paginated_queryset = self.paginate_queryset(queryset)	1✔
223	serializer = AssignmentSerializer(paginated_queryset, many=True)	1✔
224	return self.get_paginated_response(serializer.data)	1✔

uw-it-aca / canvas-analytics / 11896297726

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous