#1361

Committed 25 Apr 2024 05:39PM UTC coverage: 80.027% (+0.1%) from 79.918%

Build # #1361

Build Type

push

coveralls-python

Committed by

web-flow

Commit Message

Merge pull request #1415 from akvo/feature/1404-data-generator-jobs

Feature/1404 data generator jobs

Run Details

2590 of 3414 branches covered (75.86%)

Branch coverage included in aggregate %.

6185 of 7551 relevant lines covered (81.91%)

0.82 hits per line

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

90.59

backend/api/v1/v1_profile/views.py

# Create your views here.
import os
from typing import cast
from wsgiref.util import FileWrapper
from django.contrib.admin.sites import site
from django.core.handlers.wsgi import WSGIRequest
from django.db.models import ProtectedError, Q
from django.contrib.admin.utils import get_deleted_objects
from django.http.response import HttpResponse
from drf_spectacular.types import OpenApiTypes
from drf_spectacular.utils import (
    OpenApiParameter,
    extend_schema,
    inline_serializer,
)
from rest_framework.request import Request
from rest_framework.viewsets import ModelViewSet
from api.v1.v1_profile.models import (
    Administration,
    AdministrationAttribute,
    Entity,
    EntityData,
    Levels,
)
from api.v1.v1_profile.serializers import (
    AdministrationAttributeSerializer,
    AdministrationSerializer,
    EntityDataSerializer,
    EntitySerializer,
    DownloadAdministrationRequestSerializer,
    DownloadEntityDataRequestSerializer,
    ListEntityDataSerializer,
)
from api.v1.v1_profile.job import create_download_job
from api.v1.v1_users.models import SystemUser
from api.v1.v1_jobs.constants import JobTypes
from utils.upload_administration import generate_administration_excel
from utils.custom_helper import clean_array_param, maybe_int
from utils.default_serializers import DefaultResponseSerializer
from utils.custom_pagination import Pagination
from rest_framework.decorators import api_view, permission_classes
from rest_framework import serializers, status
from rest_framework.response import Response
from rest_framework.permissions import IsAuthenticated
from utils.email_helper import send_email, EmailTypes
from utils.custom_serializer_fields import validate_serializers_message
from utils.custom_generator import administration_csv_delete


@extend_schema(
    request=inline_serializer(
        "BatchUserComment",
        fields={
            "name": serializers.CharField(),
            "email": serializers.CharField(),
            "message": serializers.CharField(),
        },
    ),
    responses={200: DefaultResponseSerializer},
    tags=["Feedback"],
    description="Send feedback",
    summary="Send feedback",
)
@api_view(["POST"])
def send_feedback(request, version):
    name = request.data.get("name")
    email = request.data.get("email")
    message = request.data.get("message")
    # TODO:: change email
    data = {
        "send_to": ["tech.consultancy@akvo.org"],
        "subject": "Feedback from {0} <{1}>".format(name, email),
        "body": "This is feedback from {0} <{1}>. Message: {2}".format(
            name, email, message
        ),
    }
    send_email(context=data, type=EmailTypes.feedback)
    return Response(
        {"message": "Feedback was sent successfully."},
        status=status.HTTP_200_OK,
    )


@extend_schema(
    responses={200: ListEntityDataSerializer},
    tags=["Entities"],
    summary="Get list of entity data by entity type & administration",
)
@api_view(["GET"])
def list_entity_data(request, version, entity_id, administration_id):
    instance = EntityData.objects.filter(
        entity__id=entity_id, administration__id=administration_id
    ).all()
    return Response(
        ListEntityDataSerializer(instance=instance, many=True).data,
        status=status.HTTP_200_OK,
    )


@extend_schema(tags=["Administration"])
class AdministrationViewSet(ModelViewSet):
    serializer_class = AdministrationSerializer
    permission_classes = [IsAuthenticated]
    pagination_class = Pagination

    def get_queryset(self):
        queryset = Administration.objects.prefetch_related(
            "parent_administration", "attributes"
        ).all()
        search = self.request.query_params.get("search")
        parent_id = self.request.query_params.get("parent")
        level_id = self.request.query_params.get("level")
        if search:
            queryset = queryset.filter(
                Q(name__icontains=search) | Q(code__icontains=search)
            )
        if parent_id:
            try:
                parent = Administration.objects.get(id=parent_id)
                queryset = queryset.filter(
                    path__startswith=f"{parent.path or ''}{parent.id}."
                )
            except Administration.DoesNotExist:
                pass
        if level_id:
            try:
                level = Levels.objects.get(id=level_id)
                queryset = queryset.filter(level=level)
            except Levels.DoesNotExist:
                pass
        return queryset.order_by("id")

    def get_serializer(self, *args, **kwargs):
        if self.action == "list":
            kwargs.update({"compact": True})
        return super().get_serializer(*args, **kwargs)

    @extend_schema(
        parameters=[
            OpenApiParameter(
                name="parent",
                type=OpenApiTypes.NUMBER,
                location=OpenApiParameter.QUERY,
            ),
            OpenApiParameter(
                name="level",
                type=OpenApiTypes.NUMBER,
                location=OpenApiParameter.QUERY,
            ),
            OpenApiParameter(
                name="search",
                type=OpenApiTypes.STR,
                location=OpenApiParameter.QUERY,
            ),
        ]
    )
    def list(self, request, *args, **kwargs):
        return super().list(request, *args, **kwargs)

    def destroy(self, request, *args, **kwargs):
        instance = self.get_object()
        try:
            TESTING = os.environ.get("TESTING")
            administration_csv_delete(id=instance.pk, test=TESTING)
            instance.delete()
        except ProtectedError:
            _, _, _, protected = get_deleted_objects(
                [instance], cast(WSGIRequest, request), site
            )
            error = (
                f'Cannot delete "Administration: {instance}" because it is '
                "referenced by other data"
            )
            return Response(
                {"error": error, "referenced_by": protected},
                status=status.HTTP_409_CONFLICT,
            )
        return Response(status=status.HTTP_204_NO_CONTENT)


@extend_schema(tags=["Administration"])
class AdministrationAttributeViewSet(ModelViewSet):
    queryset = AdministrationAttribute.objects.order_by("id").all()
    serializer_class = AdministrationAttributeSerializer
    permission_classes = [IsAuthenticated]
    pagination_class = None


@extend_schema(tags=["Entities"])
class EntityViewSet(ModelViewSet):
    serializer_class = EntitySerializer
    permission_classes = [IsAuthenticated]
    pagination_class = Pagination

    def get_queryset(self):
        queryset = Entity.objects.all()
        search = self.request.query_params.get("search")
        if search:
            queryset = queryset.filter(name__icontains=search)
        return queryset.order_by("id")


@extend_schema(tags=["Entities"])
class EntityDataViewSet(ModelViewSet):
    serializer_class = EntityDataSerializer
    permission_classes = [IsAuthenticated]
    pagination_class = Pagination

    def get_queryset(self):
        queryset = EntityData.objects.select_related(
            "administration", "entity"
        ).all()
        search = self.request.query_params.get("search")
        adm_id = self.request.query_params.get("administration")
        entity_id = self.request.query_params.get("entity")
        if search:
            queryset = queryset.filter(
                Q(name__icontains=search) | Q(code__icontains=search)
            )
        if adm_id:
            try:
                adm_root = Administration.objects.get(id=adm_id)
                adms = Administration.objects.filter(
                    Q(path__startswith=f"{adm_root.path or ''}{adm_root.id}.")
                    | Q(id=adm_root.id)
                )
                queryset = queryset.filter(administration__in=adms)
            except Administration.DoesNotExist:
                pass
        if entity_id:
            try:
                entities = [int(e) for e in entity_id.split(",")]
                queryset = queryset.filter(entity__in=entities)
            except Entity.DoesNotExist:
                pass

        return queryset.order_by("id")

    @extend_schema(
        parameters=[
            OpenApiParameter(
                name="administration",
                type=OpenApiTypes.NUMBER,
                location=OpenApiParameter.QUERY,
            ),
            OpenApiParameter(
                name="entity",
                type=OpenApiTypes.STR,
                location=OpenApiParameter.QUERY,
            ),
            OpenApiParameter(
                name="search",
                type=OpenApiTypes.STR,
                location=OpenApiParameter.QUERY,
            ),
        ]
    )
    def list(self, request, *args, **kwargs):
        return super().list(request, *args, **kwargs)


@extend_schema(
    tags=["File"],
    summary="Export template for Administration bulk upload",
    parameters=[
        OpenApiParameter(
            name="attributes",
            type={"type": "array", "items": {"type": "number"}},
            location=OpenApiParameter.QUERY,
            explode=False,
        )
    ],
)
@api_view(["GET"])
@permission_classes([IsAuthenticated])
def export_administrations_template(request: Request, version):
    attributes = clean_array_param(
        request.query_params.get("attributes", ""), maybe_int
    )
    filepath = generate_administration_excel(
        cast(SystemUser, request.user), attributes
    )
    filename = filepath.split("/")[-1].replace(" ", "-")
    with open(filepath, "rb") as template_file:
        response = HttpResponse(
            FileWrapper(template_file),
            content_type=(
                "application/vnd.openxmlformats-officedocument"
                ".spreadsheetml.sheet"
            ),
        )
        response["Content-Disposition"] = f'attachment; filename="{filename}"'
        return response


@extend_schema(
    tags=["File"],
    summary=("Export prefilled template for Administration bulk upload"),
    parameters=[
        OpenApiParameter(
            name="attributes",
            type={"type": "array", "items": {"type": "number"}},
            location=OpenApiParameter.QUERY,
            explode=False,
        ),
        OpenApiParameter(
            name="level",
            required=False,
            type=OpenApiTypes.NUMBER,
            location=OpenApiParameter.QUERY,
        ),
    ],
)
@api_view(["GET"])
@permission_classes([IsAuthenticated])
def export_prefilled_administrations_template(request: Request, version):
    serializer = DownloadAdministrationRequestSerializer(data=request.GET)
    if not serializer.is_valid():
        return Response(
            {"message": validate_serializers_message(serializer.errors)},
            status=status.HTTP_400_BAD_REQUEST,
        )
    attributes = clean_array_param(
        request.query_params.get("attributes", ""), maybe_int
    )
    administration = request.query_params.get("administration")
    job = create_download_job(
        adm_id=administration,
        user_id=request.user.id,
        job_type=JobTypes.download_administration,
        job_info={"administration": administration, "attributes": attributes},
    )
    file_url = f"/download/file/{job.result}?type=download_administration"
    data = {
        "task_id": job.task_id,
        "file_url": file_url,
    }
    return Response(data, status=status.HTTP_200_OK)


@extend_schema(
    tags=["File"],
    summary="Export entity data",
    parameters=[
        OpenApiParameter(
            name="entity_ids",
            required=False,
            type={"entity_ids": "array", "items": {"type": "number"}},
            location=OpenApiParameter.QUERY,
            explode=False,
        ),
        OpenApiParameter(
            name="adm_id",
            required=False,
            type=OpenApiTypes.NUMBER,
            location=OpenApiParameter.QUERY,
        ),
    ],
)
@api_view(["GET"])
@permission_classes([IsAuthenticated])
def export_entity_data(request: Request, version):
    serializer = DownloadEntityDataRequestSerializer(data=request.GET)
    if not serializer.is_valid():
        return Response(
            {"message": validate_serializers_message(serializer.errors)},
            status=status.HTTP_400_BAD_REQUEST,
        )
    entity_ids = clean_array_param(
        request.query_params.get("entity_ids", ""), maybe_int
    )
    adm_id = request.query_params.get("adm_id")
    entities = Entity.objects.filter(pk__in=entity_ids).values("id", "name")
    entities = [e for e in entities]
    job = create_download_job(
        adm_id=adm_id,
        user_id=request.user.id,
        job_type=JobTypes.download_entities,
        job_info={"administration": adm_id, "entities": entities},
    )
    file_url = f"/download/file/{job.result}?type=download_entities"
    data = {
        "task_id": job.task_id,
        "file_url": file_url,
    }
    return Response(data, status=status.HTTP_200_OK)

1	# Create your views here.
2	import os	1✔
3	from typing import cast	1✔
4	from wsgiref.util import FileWrapper	1✔
5	from django.contrib.admin.sites import site	1✔
6	from django.core.handlers.wsgi import WSGIRequest	1✔
7	from django.db.models import ProtectedError, Q	1✔
8	from django.contrib.admin.utils import get_deleted_objects	1✔
9	from django.http.response import HttpResponse	1✔
10	from drf_spectacular.types import OpenApiTypes	1✔
11	from drf_spectacular.utils import (	1✔
12	OpenApiParameter,
13	extend_schema,
14	inline_serializer,
15	)
16	from rest_framework.request import Request	1✔
17	from rest_framework.viewsets import ModelViewSet	1✔
18	from api.v1.v1_profile.models import (	1✔
19	Administration,
20	AdministrationAttribute,
21	Entity,
22	EntityData,
23	Levels,
24	)
25	from api.v1.v1_profile.serializers import (	1✔
26	AdministrationAttributeSerializer,
27	AdministrationSerializer,
28	EntityDataSerializer,
29	EntitySerializer,
30	DownloadAdministrationRequestSerializer,
31	DownloadEntityDataRequestSerializer,
32	ListEntityDataSerializer,
33	)
34	from api.v1.v1_profile.job import create_download_job	1✔
35	from api.v1.v1_users.models import SystemUser	1✔
36	from api.v1.v1_jobs.constants import JobTypes	1✔
37	from utils.upload_administration import generate_administration_excel	1✔
38	from utils.custom_helper import clean_array_param, maybe_int	1✔
39	from utils.default_serializers import DefaultResponseSerializer	1✔
40	from utils.custom_pagination import Pagination	1✔
41	from rest_framework.decorators import api_view, permission_classes	1✔
42	from rest_framework import serializers, status	1✔
43	from rest_framework.response import Response	1✔
44	from rest_framework.permissions import IsAuthenticated	1✔
45	from utils.email_helper import send_email, EmailTypes	1✔
46	from utils.custom_serializer_fields import validate_serializers_message	1✔
47	from utils.custom_generator import administration_csv_delete	1✔
48
49
50	@extend_schema(	1✔
51	request=inline_serializer(
52	"BatchUserComment",
53	fields={
54	"name": serializers.CharField(),
55	"email": serializers.CharField(),
56	"message": serializers.CharField(),
57	},
58	),
59	responses={200: DefaultResponseSerializer},
60	tags=["Feedback"],
61	description="Send feedback",
62	summary="Send feedback",
63	)
64	@api_view(["POST"])	1✔
65	def send_feedback(request, version):	1✔
66	name = request.data.get("name")	1✔
67	email = request.data.get("email")	1✔
68	message = request.data.get("message")	1✔
69	# TODO:: change email
70	data = {	1✔
71	"send_to": ["tech.consultancy@akvo.org"],
72	"subject": "Feedback from {0} <{1}>".format(name, email),
73	"body": "This is feedback from {0} <{1}>. Message: {2}".format(
74	name, email, message
75	),
76	}
77	send_email(context=data, type=EmailTypes.feedback)	1✔
78	return Response(	1✔
79	{"message": "Feedback was sent successfully."},
80	status=status.HTTP_200_OK,
81	)
82
83
84	@extend_schema(	1✔
85	responses={200: ListEntityDataSerializer},
86	tags=["Entities"],
87	summary="Get list of entity data by entity type & administration",
88	)
89	@api_view(["GET"])	1✔
90	def list_entity_data(request, version, entity_id, administration_id):	1✔
91	instance = EntityData.objects.filter(	1✔
92	entity__id=entity_id, administration__id=administration_id
93	).all()
94	return Response(	1✔
95	ListEntityDataSerializer(instance=instance, many=True).data,
96	status=status.HTTP_200_OK,
97	)
98
99
100	@extend_schema(tags=["Administration"])	1✔
101	class AdministrationViewSet(ModelViewSet):	1✔
102	serializer_class = AdministrationSerializer	1✔
103	permission_classes = [IsAuthenticated]	1✔
104	pagination_class = Pagination	1✔
105
106	def get_queryset(self):	1✔
107	queryset = Administration.objects.prefetch_related(	1✔
108	"parent_administration", "attributes"
109	).all()
110	search = self.request.query_params.get("search")	1✔
111	parent_id = self.request.query_params.get("parent")	1✔
112	level_id = self.request.query_params.get("level")	1✔
113	if search:	1✔
114	queryset = queryset.filter(	1✔
115	Q(name__icontains=search) \| Q(code__icontains=search)
116	)
117	if parent_id:	1✔
118	try:	1✔
119	parent = Administration.objects.get(id=parent_id)	1✔
120	queryset = queryset.filter(	1✔
121	path__startswith=f"{parent.path or ''}{parent.id}."
122	)
123	except Administration.DoesNotExist:	×
124	pass	×
125	if level_id:	1✔
126	try:	1✔
127	level = Levels.objects.get(id=level_id)	1✔
128	queryset = queryset.filter(level=level)	1✔
129	except Levels.DoesNotExist:	×
130	pass	×
131	return queryset.order_by("id")	1✔
132
133	def get_serializer(self, args, *kwargs):	1✔
134	if self.action == "list":	1✔
135	kwargs.update({"compact": True})	1✔
136	return super().get_serializer(args, *kwargs)	1✔
137
138	@extend_schema(	1✔
139	parameters=[
140	OpenApiParameter(
141	name="parent",
142	type=OpenApiTypes.NUMBER,
143	location=OpenApiParameter.QUERY,
144	),
145	OpenApiParameter(
146	name="level",
147	type=OpenApiTypes.NUMBER,
148	location=OpenApiParameter.QUERY,
149	),
150	OpenApiParameter(
151	name="search",
152	type=OpenApiTypes.STR,
153	location=OpenApiParameter.QUERY,
154	),
155	]
156	)
157	def list(self, request, args, *kwargs):	1✔
158	return super().list(request, args, *kwargs)	1✔
159
160	def destroy(self, request, args, *kwargs):	1✔
161	instance = self.get_object()	1✔
162	try:	1✔
163	TESTING = os.environ.get("TESTING")	1✔
164	administration_csv_delete(id=instance.pk, test=TESTING)	1✔
165	instance.delete()	1✔
166	except ProtectedError:	1✔
167	_, _, _, protected = get_deleted_objects(	1✔
168	[instance], cast(WSGIRequest, request), site
169	)
170	error = (	1✔
171	f'Cannot delete "Administration: {instance}" because it is '
172	"referenced by other data"
173	)
174	return Response(	1✔
175	{"error": error, "referenced_by": protected},
176	status=status.HTTP_409_CONFLICT,
177	)
178	return Response(status=status.HTTP_204_NO_CONTENT)	1✔
179
180
181	@extend_schema(tags=["Administration"])	1✔
182	class AdministrationAttributeViewSet(ModelViewSet):	1✔
183	queryset = AdministrationAttribute.objects.order_by("id").all()	1✔
184	serializer_class = AdministrationAttributeSerializer	1✔
185	permission_classes = [IsAuthenticated]	1✔
186	pagination_class = None	1✔
187
188
189	@extend_schema(tags=["Entities"])	1✔
190	class EntityViewSet(ModelViewSet):	1✔
191	serializer_class = EntitySerializer	1✔
192	permission_classes = [IsAuthenticated]	1✔
193	pagination_class = Pagination	1✔
194
195	def get_queryset(self):	1✔
196	queryset = Entity.objects.all()	1✔
197	search = self.request.query_params.get("search")	1✔
198	if search:	1✔
199	queryset = queryset.filter(name__icontains=search)	1✔
200	return queryset.order_by("id")	1✔
201
202
203	@extend_schema(tags=["Entities"])	1✔
204	class EntityDataViewSet(ModelViewSet):	1✔
205	serializer_class = EntityDataSerializer	1✔
206	permission_classes = [IsAuthenticated]	1✔
207	pagination_class = Pagination	1✔
208
209	def get_queryset(self):	1✔
210	queryset = EntityData.objects.select_related(	1✔
211	"administration", "entity"
212	).all()
213	search = self.request.query_params.get("search")	1✔
214	adm_id = self.request.query_params.get("administration")	1✔
215	entity_id = self.request.query_params.get("entity")	1✔
216	if search:	1✔
217	queryset = queryset.filter(	1✔
218	Q(name__icontains=search) \| Q(code__icontains=search)
219	)
220	if adm_id:	1✔
221	try:	1✔
222	adm_root = Administration.objects.get(id=adm_id)	1✔
223	adms = Administration.objects.filter(	1✔
224	Q(path__startswith=f"{adm_root.path or ''}{adm_root.id}.")
225	\| Q(id=adm_root.id)
226	)
227	queryset = queryset.filter(administration__in=adms)	1✔
228	except Administration.DoesNotExist:	×
229	pass	×
230	if entity_id:	1✔
231	try:	1✔
232	entities = [int(e) for e in entity_id.split(",")]	1✔
233	queryset = queryset.filter(entity__in=entities)	1✔
234	except Entity.DoesNotExist:	×
235	pass	×
236
237	return queryset.order_by("id")	1✔
238
239	@extend_schema(	1✔
240	parameters=[
241	OpenApiParameter(
242	name="administration",
243	type=OpenApiTypes.NUMBER,
244	location=OpenApiParameter.QUERY,
245	),
246	OpenApiParameter(
247	name="entity",
248	type=OpenApiTypes.STR,
249	location=OpenApiParameter.QUERY,
250	),
251	OpenApiParameter(
252	name="search",
253	type=OpenApiTypes.STR,
254	location=OpenApiParameter.QUERY,
255	),
256	]
257	)
258	def list(self, request, args, *kwargs):	1✔
259	return super().list(request, args, *kwargs)	1✔
260
261
262	@extend_schema(	1✔
263	tags=["File"],
264	summary="Export template for Administration bulk upload",
265	parameters=[
266	OpenApiParameter(
267	name="attributes",
268	type={"type": "array", "items": {"type": "number"}},
269	location=OpenApiParameter.QUERY,
270	explode=False,
271	)
272	],
273	)
274	@api_view(["GET"])	1✔
275	@permission_classes([IsAuthenticated])	1✔
276	def export_administrations_template(request: Request, version):	1✔
277	attributes = clean_array_param(	1✔
278	request.query_params.get("attributes", ""), maybe_int
279	)
280	filepath = generate_administration_excel(	1✔
281	cast(SystemUser, request.user), attributes
282	)
283	filename = filepath.split("/")[-1].replace(" ", "-")	1✔
284	with open(filepath, "rb") as template_file:	1✔
285	response = HttpResponse(	1✔
286	FileWrapper(template_file),
287	content_type=(
288	"application/vnd.openxmlformats-officedocument"
289	".spreadsheetml.sheet"
290	),
291	)
292	response["Content-Disposition"] = f'attachment; filename="{filename}"'	1✔
293	return response	1✔
294
295
296	@extend_schema(	1✔
297	tags=["File"],
298	summary=("Export prefilled template for Administration bulk upload"),
299	parameters=[
300	OpenApiParameter(
301	name="attributes",
302	type={"type": "array", "items": {"type": "number"}},
303	location=OpenApiParameter.QUERY,
304	explode=False,
305	),
306	OpenApiParameter(
307	name="level",
308	required=False,
309	type=OpenApiTypes.NUMBER,
310	location=OpenApiParameter.QUERY,
311	),
312	],
313	)
314	@api_view(["GET"])	1✔
315	@permission_classes([IsAuthenticated])	1✔
316	def export_prefilled_administrations_template(request: Request, version):	1✔
317	serializer = DownloadAdministrationRequestSerializer(data=request.GET)	×
318	if not serializer.is_valid():	×
319	return Response(	×
320	{"message": validate_serializers_message(serializer.errors)},
321	status=status.HTTP_400_BAD_REQUEST,
322	)
323	attributes = clean_array_param(	×
324	request.query_params.get("attributes", ""), maybe_int
325	)
326	administration = request.query_params.get("administration")	×
327	job = create_download_job(	×
328	adm_id=administration,
329	user_id=request.user.id,
330	job_type=JobTypes.download_administration,
331	job_info={"administration": administration, "attributes": attributes},
332	)
333	file_url = f"/download/file/{job.result}?type=download_administration"	×
334	data = {	×
335	"task_id": job.task_id,
336	"file_url": file_url,
337	}
338	return Response(data, status=status.HTTP_200_OK)	×
339
340
341	@extend_schema(	1✔
342	tags=["File"],
343	summary="Export entity data",
344	parameters=[
345	OpenApiParameter(
346	name="entity_ids",
347	required=False,
348	type={"entity_ids": "array", "items": {"type": "number"}},
349	location=OpenApiParameter.QUERY,
350	explode=False,
351	),
352	OpenApiParameter(
353	name="adm_id",
354	required=False,
355	type=OpenApiTypes.NUMBER,
356	location=OpenApiParameter.QUERY,
357	),
358	],
359	)
360	@api_view(["GET"])	1✔
361	@permission_classes([IsAuthenticated])	1✔
362	def export_entity_data(request: Request, version):	1✔
363	serializer = DownloadEntityDataRequestSerializer(data=request.GET)	1✔
364	if not serializer.is_valid():	1✔
365	return Response(	1✔
366	{"message": validate_serializers_message(serializer.errors)},
367	status=status.HTTP_400_BAD_REQUEST,
368	)
369	entity_ids = clean_array_param(	1✔
370	request.query_params.get("entity_ids", ""), maybe_int
371	)
372	adm_id = request.query_params.get("adm_id")	1✔
373	entities = Entity.objects.filter(pk__in=entity_ids).values("id", "name")	1✔
374	entities = [e for e in entities]	1✔
375	job = create_download_job(	1✔
376	adm_id=adm_id,
377	user_id=request.user.id,
378	job_type=JobTypes.download_entities,
379	job_info={"administration": adm_id, "entities": entities},
380	)
381	file_url = f"/download/file/{job.result}?type=download_entities"	1✔
382	data = {	1✔
383	"task_id": job.task_id,
384	"file_url": file_url,
385	}
386	return Response(data, status=status.HTTP_200_OK)	1✔

akvo / rtmis / #1361

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