9888568917

Committed 11 Jul 2024 09:02AM UTC coverage: 91.858% (+0.1%) from 91.722%

Build # 9888568917

Build Type

push

github

Committed by

web-flow

Commit Message

Hide campaigns (#162)

* Add config for local environment

* Add visible flag to campaign model

* Revert "Add config for local environment"

This reverts commit d57148d30.

Run Details

54 of 54 new or added lines in 5 files covered. (100.0%)

1 existing line in 1 file now uncovered.

2121 of 2309 relevant lines covered (91.86%)

0.92 hits per line

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

99.16

/safe_locking_service/campaigns/views.py

import csv
import logging
from io import TextIOWrapper
from typing import IO, Optional

from django.contrib.admin.views.decorators import staff_member_required
from django.contrib.auth.decorators import permission_required
from django.db.models import F, Max, Sum, Window
from django.db.models.functions import Rank
from django.http import HttpRequest, HttpResponse
from django.shortcuts import get_object_or_404, redirect, render
from django.urls import reverse
from django.utils.decorators import method_decorator
from django.views.decorators.cache import cache_page

from drf_yasg import openapi
from drf_yasg.utils import swagger_auto_schema
from rest_framework import status
from rest_framework.generics import ListAPIView, RetrieveAPIView
from rest_framework.response import Response

from safe_locking_service.campaigns.models import get_campaign_leader_board_position
from safe_locking_service.campaigns.serializers import (
    CampaignLeaderBoardSerializer,
    CampaignSerializer,
    PeriodAddressSerializer,
)
from safe_locking_service.locking_events.pagination import SmallPagination

from . import tasks
from .forms import FileUploadForm
from .models import Activity, Campaign, Period


class CampaignsView(ListAPIView):
    """
    Returns a paginated list of campaigns.
    """

    pagination_class = SmallPagination
    serializer_class = CampaignSerializer

    def get_queryset(self):
        return (
            Campaign.objects.filter(visible=True)
            .prefetch_related("activity_metadata")
            .annotate(last_updated=Max("periods__end_date"))
            .order_by("-start_date", "-end_date")
        )

    @method_decorator(cache_page(1 * 60))  # 1 minute
    def list(self, request, *args, **kwargs):
        queryset = self.get_queryset()
        serializer = self.serializer_class(queryset, many=True)
        paginated_data = self.paginate_queryset(serializer.data)
        return self.get_paginated_response(paginated_data)


class RetrieveCampaignView(RetrieveAPIView):
    """
    Returns a campaign for the provided campaign_id.
    """

    serializer_class = CampaignSerializer

    def get_queryset(self, resource_id: int):
        return (
            Campaign.objects.filter(uuid=resource_id, visible=True)
            .prefetch_related("activity_metadata")
            .annotate(last_updated=Max("periods__end_date"))
        )

    @method_decorator(cache_page(1 * 60))  # 1 minute
    def get(self, request, *args, **kwargs):
        resource_id = kwargs["resource_id"]
        queryset = self.get_queryset(resource_id)
        if not queryset:
            return Response(
                status=status.HTTP_404_NOT_FOUND,
            )

        serializer = self.serializer_class(queryset[0])
        return Response(status=status.HTTP_200_OK, data=serializer.data)


csv_headers = {
    "safe_address",
    "period_start",
    "period_end",
    "total_points",
    "boost",
    "total_boosted_points",
    "points_rank",
    "boosted_points_rank",
}

logger = logging.getLogger(__name__)


# We keep the login_required decorator so that the user can log in first if they haven't done so
@staff_member_required
@permission_required("campaigns.upload_activities", raise_exception=True)
def upload_activities_view(request: HttpRequest) -> HttpResponse:
    period_slug: Optional[str] = request.GET.get("period_slug")

    if request.method == "POST":
        form = FileUploadForm(request.POST, request.FILES, period_slug=period_slug)
        if form.is_valid():
            try:
                period = form.cleaned_data["period"]
                file = TextIOWrapper(
                    request.FILES["file"].file, encoding=request.encoding
                )
                process_activity_file(period, file)
                return redirect(reverse("admin:index"))
            except Exception as e:
                logger.warning(e)
                error_message = f"Error processing CSV: {e}"
                return render(
                    request,
                    "activities/upload.html",
                    {"form": form, "error_message": error_message},
                )

    else:
        form = FileUploadForm(period_slug=period_slug)
    return render(request, "activities/upload.html", {"form": form})


def process_activity_file(period: Period, file: IO[str]) -> None:
    reader = csv.DictReader(file)

    file_headers = reader.fieldnames
    if not csv_headers.issubset(file_headers):
        raise ValueError(
            f"File does not include one or more of the following headers: {csv_headers}"
        )

    activities_list = [row for row in reader]
    tasks.process_csv_task.delay(period.id, activities_list)


class CampaignLeaderBoardView(ListAPIView):
    """
    Return the leaderboard for a provided campaign uuid
    """

    pagination_class = SmallPagination
    serializer_class = CampaignLeaderBoardSerializer

    def get_queryset(self):
        resource_id = self.kwargs["resource_id"]
        campaign = get_object_or_404(Campaign, uuid=resource_id, visible=True)

        return (
            Activity.objects.select_related("period__campaign")
            .filter(period__campaign=campaign)
            .values("address")
            .annotate(
                total_campaign_points=Sum("total_points"),
                total_campaign_boosted_points=Sum("total_boosted_points"),
                last_boost=F("total_campaign_boosted_points")
                / F("total_campaign_points"),
            )
            .order_by(F("total_campaign_boosted_points").desc())
            .annotate(
                position=Window(
                    expression=Rank(),
                    order_by=F("total_campaign_boosted_points").desc(),
                )
            )
        )

    @method_decorator(cache_page(1 * 60))  # 1 minute
    def list(self, request, *args, **kwargs):
        queryset = self.get_queryset()
        serializer = self.serializer_class(queryset, many=True)
        paginated_data = self.paginate_queryset(serializer.data)
        return self.get_paginated_response(paginated_data)


class CampaignLeaderBoardPositionView(RetrieveAPIView):
    """
    Return the leaderboard for a provided campaign uuid and address
    """

    serializer_class = CampaignLeaderBoardSerializer

    def get_queryset(self):
        resource_id = self.kwargs["resource_id"]
        address = self.kwargs["address"]
        campaign = get_object_or_404(Campaign, uuid=resource_id, visible=True)
        return get_campaign_leader_board_position(campaign.uuid, address)

    @method_decorator(cache_page(1 * 60))  # 1 minute
    def get(self, *args, **kwargs):
        queryset = self.get_queryset()
        if not queryset:
            return Response(
                status=status.HTTP_404_NOT_FOUND,
            )

        serializer = self.serializer_class(queryset)
        return Response(status=status.HTTP_200_OK, data=serializer.data)


class GetAddressPeriodsView(ListAPIView):
    pagination_class = SmallPagination
    serializer_class = PeriodAddressSerializer

    def get_queryset(self):
        resource_id = self.kwargs["resource_id"]
        address = self.request.GET.get("holder")

        campaign = get_object_or_404(Campaign, uuid=resource_id, visible=True)

        if address:
            queryset = Activity.objects.filter(
                period__campaign=campaign, address=address
            )
        else:
            queryset = Activity.objects.filter(period__campaign=campaign)

        return queryset.select_related("period").order_by("-period__start_date")

    @swagger_auto_schema(
        manual_parameters=[
            openapi.Parameter(
                "holder",
                openapi.IN_QUERY,
                description="The address of the holder to be used as a filter",
                type=openapi.TYPE_STRING,
            )
        ]
    )
    @method_decorator(cache_page(1 * 60))  # 1 minute
    def get(self, *args, **kwargs):
        queryset = self.get_queryset()
        serializer = self.serializer_class(queryset, many=True)
        paginated_data = self.paginate_queryset(serializer.data)

        return self.get_paginated_response(paginated_data)

1	import csv	1✔
2	import logging	1✔
3	from io import TextIOWrapper	1✔
4	from typing import IO, Optional	1✔
5
6	from django.contrib.admin.views.decorators import staff_member_required	1✔
7	from django.contrib.auth.decorators import permission_required	1✔
8	from django.db.models import F, Max, Sum, Window	1✔
9	from django.db.models.functions import Rank	1✔
10	from django.http import HttpRequest, HttpResponse	1✔
11	from django.shortcuts import get_object_or_404, redirect, render	1✔
12	from django.urls import reverse	1✔
13	from django.utils.decorators import method_decorator	1✔
14	from django.views.decorators.cache import cache_page	1✔
15
16	from drf_yasg import openapi	1✔
17	from drf_yasg.utils import swagger_auto_schema	1✔
18	from rest_framework import status	1✔
19	from rest_framework.generics import ListAPIView, RetrieveAPIView	1✔
20	from rest_framework.response import Response	1✔
21
22	from safe_locking_service.campaigns.models import get_campaign_leader_board_position	1✔
23	from safe_locking_service.campaigns.serializers import (	1✔
24	CampaignLeaderBoardSerializer,
25	CampaignSerializer,
26	PeriodAddressSerializer,
27	)
28	from safe_locking_service.locking_events.pagination import SmallPagination	1✔
29
30	from . import tasks	1✔
31	from .forms import FileUploadForm	1✔
32	from .models import Activity, Campaign, Period	1✔
33
34
35	class CampaignsView(ListAPIView):	1✔
36	"""
37	Returns a paginated list of campaigns.
38	"""
39
40	pagination_class = SmallPagination	1✔
41	serializer_class = CampaignSerializer	1✔
42
43	def get_queryset(self):	1✔
44	return (	1✔
45	Campaign.objects.filter(visible=True)
46	.prefetch_related("activity_metadata")
47	.annotate(last_updated=Max("periods__end_date"))
48	.order_by("-start_date", "-end_date")
49	)
50
51	@method_decorator(cache_page(1 * 60)) # 1 minute	1✔
52	def list(self, request, args, *kwargs):	1✔
53	queryset = self.get_queryset()	1✔
54	serializer = self.serializer_class(queryset, many=True)	1✔
55	paginated_data = self.paginate_queryset(serializer.data)	1✔
56	return self.get_paginated_response(paginated_data)	1✔
57
58
59	class RetrieveCampaignView(RetrieveAPIView):	1✔
60	"""
61	Returns a campaign for the provided campaign_id.
62	"""
63
64	serializer_class = CampaignSerializer	1✔
65
66	def get_queryset(self, resource_id: int):	1✔
67	return (	1✔
68	Campaign.objects.filter(uuid=resource_id, visible=True)
69	.prefetch_related("activity_metadata")
70	.annotate(last_updated=Max("periods__end_date"))
71	)
72
73	@method_decorator(cache_page(1 * 60)) # 1 minute	1✔
74	def get(self, request, args, *kwargs):	1✔
75	resource_id = kwargs["resource_id"]	1✔
76	queryset = self.get_queryset(resource_id)	1✔
77	if not queryset:	1✔
78	return Response(	1✔
79	status=status.HTTP_404_NOT_FOUND,
80	)
81
82	serializer = self.serializer_class(queryset[0])	1✔
83	return Response(status=status.HTTP_200_OK, data=serializer.data)	1✔
84
85
86	csv_headers = {	1✔
87	"safe_address",
88	"period_start",
89	"period_end",
90	"total_points",
91	"boost",
92	"total_boosted_points",
93	"points_rank",
94	"boosted_points_rank",
95	}
96
97	logger = logging.getLogger(__name__)	1✔
98
99
100	# We keep the login_required decorator so that the user can log in first if they haven't done so
101	@staff_member_required	1✔
102	@permission_required("campaigns.upload_activities", raise_exception=True)	1✔
103	def upload_activities_view(request: HttpRequest) -> HttpResponse:	1✔
104	period_slug: Optional[str] = request.GET.get("period_slug")	1✔
105
106	if request.method == "POST":	1✔
107	form = FileUploadForm(request.POST, request.FILES, period_slug=period_slug)	1✔
108	if form.is_valid():	1✔
109	try:	1✔
110	period = form.cleaned_data["period"]	1✔
111	file = TextIOWrapper(	1✔
112	request.FILES["file"].file, encoding=request.encoding
113	)
114	process_activity_file(period, file)	1✔
115	return redirect(reverse("admin:index"))	1✔
116	except Exception as e:	1✔
117	logger.warning(e)	1✔
118	error_message = f"Error processing CSV: {e}"	1✔
119	return render(	1✔
120	request,
121	"activities/upload.html",
122	{"form": form, "error_message": error_message},
123	)
124
125	else:
126	form = FileUploadForm(period_slug=period_slug)	1✔
127	return render(request, "activities/upload.html", {"form": form})	1✔
128
129
130	def process_activity_file(period: Period, file: IO[str]) -> None:	1✔
131	reader = csv.DictReader(file)	1✔
132
133	file_headers = reader.fieldnames	1✔
134	if not csv_headers.issubset(file_headers):	1✔
135	raise ValueError(	1✔
136	f"File does not include one or more of the following headers: {csv_headers}"
137	)
138
139	activities_list = [row for row in reader]	1✔
140	tasks.process_csv_task.delay(period.id, activities_list)	1✔
141
142
143	class CampaignLeaderBoardView(ListAPIView):	1✔
144	"""
145	Return the leaderboard for a provided campaign uuid
146	"""
147
148	pagination_class = SmallPagination	1✔
149	serializer_class = CampaignLeaderBoardSerializer	1✔
150
151	def get_queryset(self):	1✔
152	resource_id = self.kwargs["resource_id"]	1✔
153	campaign = get_object_or_404(Campaign, uuid=resource_id, visible=True)	1✔
154
155	return (	1✔
156	Activity.objects.select_related("period__campaign")
157	.filter(period__campaign=campaign)
158	.values("address")
159	.annotate(
160	total_campaign_points=Sum("total_points"),
161	total_campaign_boosted_points=Sum("total_boosted_points"),
162	last_boost=F("total_campaign_boosted_points")
163	/ F("total_campaign_points"),
164	)
165	.order_by(F("total_campaign_boosted_points").desc())
166	.annotate(
167	position=Window(
168	expression=Rank(),
169	order_by=F("total_campaign_boosted_points").desc(),
170	)
171	)
172	)
173
174	@method_decorator(cache_page(1 * 60)) # 1 minute	1✔
175	def list(self, request, args, *kwargs):	1✔
176	queryset = self.get_queryset()	1✔
177	serializer = self.serializer_class(queryset, many=True)	1✔
178	paginated_data = self.paginate_queryset(serializer.data)	1✔
179	return self.get_paginated_response(paginated_data)	1✔
180
181
182	class CampaignLeaderBoardPositionView(RetrieveAPIView):	1✔
183	"""
184	Return the leaderboard for a provided campaign uuid and address
185	"""
186
187	serializer_class = CampaignLeaderBoardSerializer	1✔
188
189	def get_queryset(self):	1✔
190	resource_id = self.kwargs["resource_id"]	1✔
191	address = self.kwargs["address"]	1✔
192	campaign = get_object_or_404(Campaign, uuid=resource_id, visible=True)	1✔
193	return get_campaign_leader_board_position(campaign.uuid, address)	1✔
194
195	@method_decorator(cache_page(1 * 60)) # 1 minute	1✔
196	def get(self, args, *kwargs):	1✔
197	queryset = self.get_queryset()	1✔
198	if not queryset:	1✔
UNCOV 199	return Response(	×
200	status=status.HTTP_404_NOT_FOUND,
201	)
202
203	serializer = self.serializer_class(queryset)	1✔
204	return Response(status=status.HTTP_200_OK, data=serializer.data)	1✔
205
206
207	class GetAddressPeriodsView(ListAPIView):	1✔
208	pagination_class = SmallPagination	1✔
209	serializer_class = PeriodAddressSerializer	1✔
210
211	def get_queryset(self):	1✔
212	resource_id = self.kwargs["resource_id"]	1✔
213	address = self.request.GET.get("holder")	1✔
214
215	campaign = get_object_or_404(Campaign, uuid=resource_id, visible=True)	1✔
216
217	if address:	1✔
218	queryset = Activity.objects.filter(	1✔
219	period__campaign=campaign, address=address
220	)
221	else:
222	queryset = Activity.objects.filter(period__campaign=campaign)	1✔
223
224	return queryset.select_related("period").order_by("-period__start_date")	1✔
225
226	@swagger_auto_schema(	1✔
227	manual_parameters=[
228	openapi.Parameter(
229	"holder",
230	openapi.IN_QUERY,
231	description="The address of the holder to be used as a filter",
232	type=openapi.TYPE_STRING,
233	)
234	]
235	)
236	@method_decorator(cache_page(1 * 60)) # 1 minute	1✔
237	def get(self, args, *kwargs):	1✔
238	queryset = self.get_queryset()	1✔
239	serializer = self.serializer_class(queryset, many=True)	1✔
240	paginated_data = self.paginate_queryset(serializer.data)	1✔
241
242	return self.get_paginated_response(paginated_data)	1✔

safe-global / safe-locking-service / 9888568917

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