#1842

Committed 27 Jun 2024 11:05AM UTC coverage: 84.738% (-0.1%) from 84.867%

Build # #1842

Build Type

push

coveralls-python

Committed by

ifirmawan

Commit Message

[#1050] Update mobile version to 4.0.10

Run Details

3062 of 3747 branches covered (81.72%)

Branch coverage included in aggregate %.

6849 of 7949 relevant lines covered (86.16%)

0.86 hits per line

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

76.7

backend/api/v1/v1_categories/views.py

import pandas as pd
from io import StringIO
from django.http import Http404
from drf_spectacular.types import OpenApiTypes
from drf_spectacular.utils import (
    extend_schema,
    inline_serializer,
    OpenApiParameter,
)
from django.http import HttpResponse
from rest_framework import serializers, status
from rest_framework.decorators import api_view
from rest_framework.response import Response
from utils.custom_serializer_fields import validate_serializers_message
from utils.custom_pagination import Pagination

from api.v1.v1_categories.functions import get_category_results, get_category
from api.v1.v1_categories.models import DataCategory
from api.v1.v1_data.models import FormData, Answers
from api.v1.v1_forms.models import Questions
from api.v1.v1_categories.serializers import (
    ListRawDataSerializer,
    ListRawDataAnswerSerializer,
    ListCsvDataAnswerSerializer,
)

from api.v1.v1_data.functions import get_cache, create_cache

from api.v1.v1_data.serializers import ListFormDataRequestSerializer


@extend_schema(
    description="""
    Get datapoints with computed category
    """,
    responses={
        (200, "application/json"): inline_serializer(
            "ListDataCategorizedPaginated",
            fields={
                "current": serializers.IntegerField(),
                "total": serializers.IntegerField(),
                "total_page": serializers.IntegerField(),
                "data": inline_serializer(
                    "ListDataCategorized",
                    fields={
                        "id": serializers.IntegerField(),
                        "categories": inline_serializer(
                            "ListDataCategorySerializer",
                            fields={"Sanitation": serializers.CharField()},
                        ),
                    },
                ),
            },
            many=True,
        ),
    },
    parameters=[
        OpenApiParameter(
            name="page",
            required=True,
            type=OpenApiTypes.NUMBER,
            location=OpenApiParameter.QUERY,
        ),
    ],
    tags=["Data Categories"],
    summary="Get datapoints with computed category",
)
@api_view(["GET"])
def get_data_with_category(request, version, form_id):
    queryset = FormData.objects.filter(form_id=form_id).values_list(
        "pk", flat=True
    )
    if not len(queryset):
        raise Http404("DataCategory does not exist")
    paginator = Pagination()
    instance = paginator.paginate_queryset(queryset, request)
    categories = DataCategory.objects.filter(data_id__in=instance)
    results = get_category_results(categories)
    return paginator.get_paginated_response(results)


@extend_schema(
    description="""
    Get Multi-purpose schema of datapoints to use with third party applications
    """,
    responses={200: ListRawDataSerializer(many=True)},
    parameters=[
        OpenApiParameter(
            name="page",
            required=True,
            type=OpenApiTypes.NUMBER,
            location=OpenApiParameter.QUERY,
        ),
        OpenApiParameter(
            name="questions",
            required=False,
            type={"type": "array", "items": {"type": "number"}},
            location=OpenApiParameter.QUERY,
        ),
    ],
    tags=["Data Categories"],
    summary="Get Raw data points",
)
@api_view(["GET"])
# @permission_classes([IsAuthenticated])
def get_raw_data_point(request, version, form_id):
    serializer = ListFormDataRequestSerializer(data=request.GET)
    if not serializer.is_valid():
        return Response(
            {"message": validate_serializers_message(serializer.errors)},
            status=status.HTTP_400_BAD_REQUEST,
        )
    instances = FormData.objects.filter(form_id=form_id).order_by("-created")
    paginator = Pagination()
    page = paginator.paginate_queryset(instances, request)
    data = ListRawDataSerializer(
        instance=page,
        context={"questions": serializer.validated_data.get("questions")},
        many=True,
    ).data
    filter_data = {}
    if request.GET.get("questions"):
        filter_data["question_id__in"] = request.GET.getlist("questions")
    categories = DataCategory.objects.filter(
        form_id=form_id, data_id__in=[d["id"] for d in data]
    ).all()
    categories = get_category_results(categories)
    for d in data:
        category = list(filter(lambda x: x["id"] == d["id"], categories))
        filter_data["data_id"] = d["id"]
        answers = Answers.objects.filter(**filter_data).all()
        answers = ListRawDataAnswerSerializer(instance=answers, many=True).data
        data_answers = {}
        for a in answers:
            data_answers.update({a["question"]: a["value"]})
        if category:
            d.update(
                {"data": data_answers, "categories": category[0]["category"]}
            )
        else:
            d.update({"data": data_answers, "categories": {}})
    return paginator.get_paginated_response(data)


@extend_schema(
    description="""
    Get PowerBI schema of datapoints to use with Power BI
    """,
    parameters=[
        OpenApiParameter(
            name="cache",
            default="test",
            required=False,
            type=OpenApiTypes.STR,
            location=OpenApiParameter.QUERY,
        ),
    ],
    responses={200: ListRawDataSerializer(many=True)},
    tags=["Data Categories"],
    summary="Get Power BI data points",
)
@api_view(["GET"])
# @permission_classes([IsAuthenticated])
def get_power_bi_data(request, version, form_id):
    cache_name = request.GET.get("cache")
    if cache_name:
        cache_name = f"power_bi-{cache_name}"
        cache_data = get_cache(cache_name)
        if cache_data:
            return Response(cache_data, status=status.HTTP_200_OK)
    instances = (
        FormData.objects.filter(form_id=form_id).order_by("-created").all()
    )
    data = ListRawDataSerializer(
        instance=instances,
        many=True,
    ).data
    categories = DataCategory.objects.filter(
        form_id=form_id, data_id__in=[d["id"] for d in data]
    ).all()
    categories = get_category_results(categories)
    for d in data:
        category = list(filter(lambda x: x["id"] == d["id"], categories))
        answers = Answers.objects.filter(data_id=d["id"]).all()
        answers = ListRawDataAnswerSerializer(instance=answers, many=True).data
        data_answers = {}
        for a in answers:
            data_answers.update({a["question"]: a["value"]})
        if category:
            d.update(
                {"data": data_answers, "categories": category[0]["category"]}
            )
        else:
            d.update({"data": data_answers, "categories": {}})
    if cache_name:
        create_cache(cache_name, data)
    return Response(
        data,
        status=status.HTTP_200_OK,
    )


def generate_data(instances):
    for instance in instances:
        data = ListRawDataSerializer(instance=instance).data
        answers = Answers.objects.filter(data_id=data["id"]).all()
        answers = ListCsvDataAnswerSerializer(instance=answers, many=True).data
        data.update({a["question"]: a["value"] for a in answers})
        categories = DataCategory.objects.filter(data_id=data["id"]).all()
        if categories:
            for c in categories:
                category = get_category(c.options)
                if category:
                    data.update({c.name: category})
        yield data


@extend_schema(
    description="""
    Get csv datapoints along with JMP data
    """,
    parameters=[
        OpenApiParameter(
            name="cache",
            default="test",
            required=False,
            type=OpenApiTypes.STR,
            location=OpenApiParameter.QUERY,
        ),
    ],
    tags=["Data Categories"],
    summary="Get csv datapoints along with JMP data",
)
@api_view(["GET"])
# @permission_classes([IsAuthenticated])
def get_raw_csv_data(request, version, form_id):
    cache_name = request.GET.get("cache")
    if cache_name:
        cache_name = f"power_bi-csv-{cache_name}"
        cache_data = get_cache(cache_name)
        if cache_data:
            response = HttpResponse(cache_data, content_type="text/csv")
            response["Content-Disposition"] = 'attachment; filename="data.csv"'
            return response
    instances = (
        FormData.objects.filter(form_id=form_id).order_by("-created").all()
    )
    data = generate_data(instances)
    df = pd.DataFrame(data)
    questions = Questions.objects.filter(form_id=form_id).all()
    column_names = {}
    for question in questions:
        if question.id not in list(df):
            df[f"{question.id}|{question.name}"] = ""
        else:
            column_names.update(
                {question.id: f"{question.id}|{question.name}"}
            )
    df = df.rename(columns=column_names)
    csv_data = StringIO()
    df.to_csv(csv_data, sep=",", index=False)
    csv_text = csv_data.getvalue()
    if cache_name:
        create_cache(cache_name, csv_text)
    response = HttpResponse(csv_text, content_type="text/csv")
    response["Content-Disposition"] = 'attachment; filename="data.csv"'
    return response

1	import pandas as pd	1✔
2	from io import StringIO	1✔
3	from django.http import Http404	1✔
4	from drf_spectacular.types import OpenApiTypes	1✔
5	from drf_spectacular.utils import (	1✔
6	extend_schema,
7	inline_serializer,
8	OpenApiParameter,
9	)
10	from django.http import HttpResponse	1✔
11	from rest_framework import serializers, status	1✔
12	from rest_framework.decorators import api_view	1✔
13	from rest_framework.response import Response	1✔
14	from utils.custom_serializer_fields import validate_serializers_message	1✔
15	from utils.custom_pagination import Pagination	1✔
16
17	from api.v1.v1_categories.functions import get_category_results, get_category	1✔
18	from api.v1.v1_categories.models import DataCategory	1✔
19	from api.v1.v1_data.models import FormData, Answers	1✔
20	from api.v1.v1_forms.models import Questions	1✔
21	from api.v1.v1_categories.serializers import (	1✔
22	ListRawDataSerializer,
23	ListRawDataAnswerSerializer,
24	ListCsvDataAnswerSerializer,
25	)
26
27	from api.v1.v1_data.functions import get_cache, create_cache	1✔
28
29	from api.v1.v1_data.serializers import ListFormDataRequestSerializer	1✔
30
31
32	@extend_schema(	1✔
33	description="""
34	Get datapoints with computed category
35	""",
36	responses={
37	(200, "application/json"): inline_serializer(
38	"ListDataCategorizedPaginated",
39	fields={
40	"current": serializers.IntegerField(),
41	"total": serializers.IntegerField(),
42	"total_page": serializers.IntegerField(),
43	"data": inline_serializer(
44	"ListDataCategorized",
45	fields={
46	"id": serializers.IntegerField(),
47	"categories": inline_serializer(
48	"ListDataCategorySerializer",
49	fields={"Sanitation": serializers.CharField()},
50	),
51	},
52	),
53	},
54	many=True,
55	),
56	},
57	parameters=[
58	OpenApiParameter(
59	name="page",
60	required=True,
61	type=OpenApiTypes.NUMBER,
62	location=OpenApiParameter.QUERY,
63	),
64	],
65	tags=["Data Categories"],
66	summary="Get datapoints with computed category",
67	)
68	@api_view(["GET"])	1✔
69	def get_data_with_category(request, version, form_id):	1✔
70	queryset = FormData.objects.filter(form_id=form_id).values_list(	×
71	"pk", flat=True
72	)
73	if not len(queryset):	×
74	raise Http404("DataCategory does not exist")	×
75	paginator = Pagination()	×
76	instance = paginator.paginate_queryset(queryset, request)	×
77	categories = DataCategory.objects.filter(data_id__in=instance)	×
78	results = get_category_results(categories)	×
79	return paginator.get_paginated_response(results)	×
80
81
82	@extend_schema(	1✔
83	description="""
84	Get Multi-purpose schema of datapoints to use with third party applications
85	""",
86	responses={200: ListRawDataSerializer(many=True)},
87	parameters=[
88	OpenApiParameter(
89	name="page",
90	required=True,
91	type=OpenApiTypes.NUMBER,
92	location=OpenApiParameter.QUERY,
93	),
94	OpenApiParameter(
95	name="questions",
96	required=False,
97	type={"type": "array", "items": {"type": "number"}},
98	location=OpenApiParameter.QUERY,
99	),
100	],
101	tags=["Data Categories"],
102	summary="Get Raw data points",
103	)
104	@api_view(["GET"])	1✔
105	# @permission_classes([IsAuthenticated])
106	def get_raw_data_point(request, version, form_id):	1✔
107	serializer = ListFormDataRequestSerializer(data=request.GET)	1✔
108	if not serializer.is_valid():	1!
109	return Response(	×
110	{"message": validate_serializers_message(serializer.errors)},
111	status=status.HTTP_400_BAD_REQUEST,
112	)
113	instances = FormData.objects.filter(form_id=form_id).order_by("-created")	1✔
114	paginator = Pagination()	1✔
115	page = paginator.paginate_queryset(instances, request)	1✔
116	data = ListRawDataSerializer(	1✔
117	instance=page,
118	context={"questions": serializer.validated_data.get("questions")},
119	many=True,
120	).data
121	filter_data = {}	1✔
122	if request.GET.get("questions"):	1✔
123	filter_data["question_id__in"] = request.GET.getlist("questions")	1✔
124	categories = DataCategory.objects.filter(	1✔
125	form_id=form_id, data_id__in=[d["id"] for d in data]
126	).all()
127	categories = get_category_results(categories)	1✔
128	for d in data:	1✔
129	category = list(filter(lambda x: x["id"] == d["id"], categories))	1✔
130	filter_data["data_id"] = d["id"]	1✔
131	answers = Answers.objects.filter(**filter_data).all()	1✔
132	answers = ListRawDataAnswerSerializer(instance=answers, many=True).data	1✔
133	data_answers = {}	1✔
134	for a in answers:	1✔
135	data_answers.update({a["question"]: a["value"]})	1✔
136	if category:	1!
137	d.update(	1✔
138	{"data": data_answers, "categories": category[0]["category"]}
139	)
140	else:
141	d.update({"data": data_answers, "categories": {}})	×
142	return paginator.get_paginated_response(data)	1✔
143
144
145	@extend_schema(	1✔
146	description="""
147	Get PowerBI schema of datapoints to use with Power BI
148	""",
149	parameters=[
150	OpenApiParameter(
151	name="cache",
152	default="test",
153	required=False,
154	type=OpenApiTypes.STR,
155	location=OpenApiParameter.QUERY,
156	),
157	],
158	responses={200: ListRawDataSerializer(many=True)},
159	tags=["Data Categories"],
160	summary="Get Power BI data points",
161	)
162	@api_view(["GET"])	1✔
163	# @permission_classes([IsAuthenticated])
164	def get_power_bi_data(request, version, form_id):	1✔
165	cache_name = request.GET.get("cache")	1✔
166	if cache_name:	1!
167	cache_name = f"power_bi-{cache_name}"	×
168	cache_data = get_cache(cache_name)	×
169	if cache_data:	×
170	return Response(cache_data, status=status.HTTP_200_OK)	×
171	instances = (	1✔
172	FormData.objects.filter(form_id=form_id).order_by("-created").all()
173	)
174	data = ListRawDataSerializer(	1✔
175	instance=instances,
176	many=True,
177	).data
178	categories = DataCategory.objects.filter(	1✔
179	form_id=form_id, data_id__in=[d["id"] for d in data]
180	).all()
181	categories = get_category_results(categories)	1✔
182	for d in data:	1✔
183	category = list(filter(lambda x: x["id"] == d["id"], categories))	1✔
184	answers = Answers.objects.filter(data_id=d["id"]).all()	1✔
185	answers = ListRawDataAnswerSerializer(instance=answers, many=True).data	1✔
186	data_answers = {}	1✔
187	for a in answers:	1✔
188	data_answers.update({a["question"]: a["value"]})	1✔
189	if category:	1!
190	d.update(	1✔
191	{"data": data_answers, "categories": category[0]["category"]}
192	)
193	else:
194	d.update({"data": data_answers, "categories": {}})	×
195	if cache_name:	1!
196	create_cache(cache_name, data)	×
197	return Response(	1✔
198	data,
199	status=status.HTTP_200_OK,
200	)
201
202
203	def generate_data(instances):	1✔
204	for instance in instances:	1✔
205	data = ListRawDataSerializer(instance=instance).data	1✔
206	answers = Answers.objects.filter(data_id=data["id"]).all()	1✔
207	answers = ListCsvDataAnswerSerializer(instance=answers, many=True).data	1✔
208	data.update({a["question"]: a["value"] for a in answers})	1✔
209	categories = DataCategory.objects.filter(data_id=data["id"]).all()	1✔
210	if categories:	1!
211	for c in categories:	1✔
212	category = get_category(c.options)	1✔
213	if category:	1!
214	data.update({c.name: category})	×
215	yield data	1✔
216
217
218	@extend_schema(	1✔
219	description="""
220	Get csv datapoints along with JMP data
221	""",
222	parameters=[
223	OpenApiParameter(
224	name="cache",
225	default="test",
226	required=False,
227	type=OpenApiTypes.STR,
228	location=OpenApiParameter.QUERY,
229	),
230	],
231	tags=["Data Categories"],
232	summary="Get csv datapoints along with JMP data",
233	)
234	@api_view(["GET"])	1✔
235	# @permission_classes([IsAuthenticated])
236	def get_raw_csv_data(request, version, form_id):	1✔
237	cache_name = request.GET.get("cache")	1✔
238	if cache_name:	1!
239	cache_name = f"power_bi-csv-{cache_name}"	×
240	cache_data = get_cache(cache_name)	×
241	if cache_data:	×
242	response = HttpResponse(cache_data, content_type="text/csv")	×
243	response["Content-Disposition"] = 'attachment; filename="data.csv"'	×
244	return response	×
245	instances = (	1✔
246	FormData.objects.filter(form_id=form_id).order_by("-created").all()
247	)
248	data = generate_data(instances)	1✔
249	df = pd.DataFrame(data)	1✔
250	questions = Questions.objects.filter(form_id=form_id).all()	1✔
251	column_names = {}	1✔
252	for question in questions:	1✔
253	if question.id not in list(df):	1!
254	df[f"{question.id}\|{question.name}"] = ""	×
255	else:
256	column_names.update(	1✔
257	{question.id: f"{question.id}\|{question.name}"}
258	)
259	df = df.rename(columns=column_names)	1✔
260	csv_data = StringIO()	1✔
261	df.to_csv(csv_data, sep=",", index=False)	1✔
262	csv_text = csv_data.getvalue()	1✔
263	if cache_name:	1!
264	create_cache(cache_name, csv_text)	×
265	response = HttpResponse(csv_text, content_type="text/csv")	1✔
266	response["Content-Disposition"] = 'attachment; filename="data.csv"'	1✔
267	return response	1✔

akvo / rtmis / #1842

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