#1313

Committed 24 Apr 2024 10:13PM UTC coverage: 79.932% (+0.2%) from 79.77%

Build # #1313

Build Type

push

coveralls-python

Committed by

web-flow

Commit Message

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

Feature/1404 data generator jobs

Run Details

2570 of 3393 branches covered (75.74%)

Branch coverage included in aggregate %.

6153 of 7520 relevant lines covered (81.82%)

0.82 hits per line

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

96.45

backend/api/v1/v1_data/models.py

from django.db import models

# Create your models here.
from api.v1.v1_data.constants import DataApprovalStatus
from api.v1.v1_forms.constants import (
    QuestionTypes,
    SubmissionTypes,
)
from api.v1.v1_forms.models import Forms, Questions
from api.v1.v1_profile.models import Administration, Levels
from api.v1.v1_users.models import SystemUser
from utils.soft_deletes_model import SoftDeletes
import os
import uuid
import json
from utils import storage


class FormData(models.Model):
    parent = models.ForeignKey(
        "self",
        on_delete=models.CASCADE,
        related_name="children",
        null=True,
        blank=True,
    )
    name = models.TextField()
    form = models.ForeignKey(
        to=Forms, on_delete=models.CASCADE, related_name="form_form_data"
    )
    administration = models.ForeignKey(
        to=Administration,
        on_delete=models.PROTECT,
        related_name="administration_form_data",
    )
    geo = models.JSONField(null=True, default=None)
    uuid = models.CharField(max_length=255, default=uuid.uuid4, null=True)
    submission_type = models.IntegerField(
        choices=SubmissionTypes.FieldStr.items(),
        default=SubmissionTypes.registration,
    )
    created_by = models.ForeignKey(
        to=SystemUser,
        on_delete=models.CASCADE,
        related_name="form_data_created",
    )
    updated_by = models.ForeignKey(
        to=SystemUser,
        on_delete=models.CASCADE,
        related_name="form_data_updated",
        default=None,
        null=True,
    )
    created = models.DateTimeField(auto_now_add=True)
    updated = models.DateTimeField(default=None, null=True)

    def __str__(self):
        return self.name

    @property
    def to_data_frame(self):
        administration = self.administration
        data = {
            "id": self.id,
            "datapoint_name": self.name,
            "administration": administration.administration_column
            if administration
            else None,
            "uuid": self.uuid,
            "geolocation": f"{self.geo[0]}, {self.geo[1]}"
            if self.geo
            else None,
            "created_by": self.created_by.get_full_name(),
            "updated_by": self.updated_by.get_full_name()
            if self.updated_by
            else None,
            "created_at": self.created.strftime("%B %d, %Y %I:%M %p"),
            "updated_at": self.updated.strftime("%B %d, %Y %I:%M %p")
            if self.updated
            else None,
            "submission_type":
            SubmissionTypes().FieldStr[self.submission_type],
        }
        for a in self.data_answer.order_by(
            "question__question_group_id", "question__order"
        ).all():
            data.update(a.to_data_frame)
        return data

    @property
    def save_to_file(self):
        data = {
            "id": self.id,
            "datapoint_name": self.name,
            "administration": self.administration.id
            if self.administration
            else None,
            "uuid": str(self.uuid),
            "geolocation": self.geo,
            "submission_type": self.submission_type,
        }
        answers = {}
        for a in self.data_answer.order_by(
            "question__question_group_id", "question__order"
        ).all():
            answers.update(a.to_key)
        data.update({"answers": answers})
        json_data = json.dumps(data)
        file_name = f"{str(self.uuid)}.json"
        # write to json file
        with open(file_name, "w") as f:
            f.write(json_data)
        storage.upload(file=file_name, folder="datapoints")
        # delete file
        os.remove(file_name)
        return data

    @property
    def loc(self):
        return self.administration.name

    class Meta:
        db_table = "data"


class PendingDataBatch(models.Model):
    form = models.ForeignKey(
        to=Forms, on_delete=models.CASCADE, related_name="form_batch_data"
    )
    administration = models.ForeignKey(
        to=Administration,
        on_delete=models.PROTECT,
        related_name="administration_pending_data_batch",
    )
    user = models.ForeignKey(
        to=SystemUser,
        on_delete=models.CASCADE,
        related_name="user_pending_data_batch",
    )
    name = models.TextField()
    uuid = models.UUIDField(default=None, null=True)
    file = models.URLField(default=None, null=True)
    approved = models.BooleanField(default=False)
    created = models.DateTimeField(auto_now_add=True)
    updated = models.DateTimeField(default=None, null=True)

    def __str__(self):
        return self.name

    class Meta:
        db_table = "batch"


class PendingDataBatchComments(models.Model):
    batch = models.ForeignKey(
        to=PendingDataBatch,
        on_delete=models.CASCADE,
        related_name="batch_batch_comment",
    )
    user = models.ForeignKey(
        to=SystemUser,
        on_delete=models.CASCADE,
        related_name="user_batch_comment",
    )
    comment = models.TextField()
    created = models.DateTimeField(auto_now_add=True)

    def __str__(self):
        return self.comment

    class Meta:
        db_table = "batch_comment"


class PendingFormData(SoftDeletes):
    name = models.TextField()
    form = models.ForeignKey(
        to=Forms,
        on_delete=models.CASCADE,
        related_name="pending_form_form_data",
    )
    data = models.ForeignKey(
        to=FormData,
        on_delete=models.CASCADE,
        related_name="pending_data_form_data",
        default=None,
        null=True,
    )
    submission_type = models.IntegerField(
        choices=SubmissionTypes.FieldStr.items(),
        default=SubmissionTypes.registration,
    )
    administration = models.ForeignKey(
        to=Administration,
        on_delete=models.PROTECT,
        related_name="administration_pending_form_data",
    )  # noqa
    geo = models.JSONField(null=True, default=None)
    uuid = models.CharField(max_length=255, default=uuid.uuid4, null=True)
    batch = models.ForeignKey(
        to=PendingDataBatch,
        on_delete=models.SET_NULL,
        default=None,
        null=True,
        related_name="batch_pending_data_batch",
    )
    created_by = models.ForeignKey(
        to=SystemUser,
        on_delete=models.CASCADE,
        related_name="pending_form_data_created",
    )
    updated_by = models.ForeignKey(
        to=SystemUser,
        on_delete=models.PROTECT,
        related_name="pending_form_data_updated",
        default=None,
        null=True,
    )
    created = models.DateTimeField(auto_now_add=True)
    updated = models.DateTimeField(default=None, null=True)
    duration = models.IntegerField(default=0)
    submitter = models.CharField(max_length=255, default=None, null=True)

    def __str__(self):
        return self.name

    class Meta:
        db_table = "pending_data"


class PendingDataApproval(models.Model):
    batch = models.ForeignKey(
        to=PendingDataBatch,
        on_delete=models.CASCADE,
        related_name="batch_approval",
    )
    user = models.ForeignKey(
        to=SystemUser,
        on_delete=models.CASCADE,
        related_name="user_assigned_pending_data",
    )
    level = models.ForeignKey(
        to=Levels,
        on_delete=models.CASCADE,
        related_name="level_assigned_pending_data",
    )
    status = models.IntegerField(
        choices=DataApprovalStatus.FieldStr.items(),
        default=DataApprovalStatus.pending,
    )

    def __str__(self):
        return self.user.email

    class Meta:
        db_table = "pending_data_approval"


class PendingAnswers(models.Model):
    pending_data = models.ForeignKey(
        to=PendingFormData,
        on_delete=models.CASCADE,
        related_name="pending_data_answer",
    )
    question = models.ForeignKey(
        to=Questions,
        on_delete=models.CASCADE,
        related_name="question_pending_answer",
    )
    name = models.TextField(null=True, default=None)
    value = models.FloatField(null=True, default=None)
    options = models.JSONField(default=None, null=True)
    created_by = models.ForeignKey(
        to=SystemUser,
        on_delete=models.CASCADE,
        related_name="pending_answer_created",
    )
    created = models.DateTimeField(auto_now_add=True)
    updated = models.DateTimeField(default=None, null=True)

    def __str__(self):
        return str(self.id)

    class Meta:
        db_table = "pending_answer"


class PendingAnswerHistory(models.Model):
    pending_data = models.ForeignKey(
        to=PendingFormData,
        on_delete=models.CASCADE,
        related_name="pending_data_answer_history",
    )
    question = models.ForeignKey(
        to=Questions,
        on_delete=models.CASCADE,
        related_name="question_pending_answer_history",
    )
    name = models.TextField(null=True, default=None)
    value = models.FloatField(null=True, default=None)
    options = models.JSONField(default=None, null=True)
    created_by = models.ForeignKey(
        to=SystemUser,
        on_delete=models.PROTECT,
        related_name="pending_answer_history_created",
    )
    created = models.DateTimeField(auto_now_add=True)
    updated = models.DateTimeField(default=None, null=True)

    def __str__(self):
        return self.name

    class Meta:
        db_table = "pending_answer_history"


class Answers(models.Model):
    data = models.ForeignKey(
        to=FormData, on_delete=models.CASCADE, related_name="data_answer"
    )
    question = models.ForeignKey(
        to=Questions, on_delete=models.CASCADE, related_name="question_answer"
    )
    name = models.TextField(null=True, default=None)
    value = models.FloatField(null=True, default=None)
    options = models.JSONField(default=None, null=True)
    created_by = models.ForeignKey(
        to=SystemUser, on_delete=models.CASCADE, related_name="answer_created"
    )
    created = models.DateTimeField(auto_now_add=True)
    updated = models.DateTimeField(default=None, null=True)

    def __str__(self):
        return self.data.name

    @property
    def to_data_frame(self) -> dict:
        q = self.question
        qname = f"{self.question.name}"
        if q.type in [
            QuestionTypes.geo,
            QuestionTypes.option,
            QuestionTypes.multiple_option,
        ]:
            answer = "|".join(map(str, self.options))
        elif q.type in [
            QuestionTypes.text,
            QuestionTypes.photo,
            QuestionTypes.date,
            QuestionTypes.autofield,
        ]:
            answer = self.name
        elif q.type == QuestionTypes.administration:
            answer = Administration.objects.filter(pk=self.value).first()
            if answer:
                answer = answer.administration_column
        else:
            answer = self.value
        return {qname: answer}

    @property
    def to_key(self) -> dict:
        q = self.question
        if q.type in [
            QuestionTypes.geo,
            QuestionTypes.option,
            QuestionTypes.multiple_option,
        ]:
            answer = self.options
        elif q.type in [
            QuestionTypes.text,
            QuestionTypes.photo,
            QuestionTypes.date,
            QuestionTypes.cascade,
        ]:
            answer = self.name
        else:
            answer = self.value
        return {q.id: answer}

    class Meta:
        db_table = "answer"


class AnswerHistory(models.Model):
    data = models.ForeignKey(
        to=FormData,
        on_delete=models.CASCADE,
        related_name="data_answer_history",
    )
    question = models.ForeignKey(
        to=Questions,
        on_delete=models.CASCADE,
        related_name="question_answer_history",
    )
    name = models.TextField(null=True, default=None)
    value = models.FloatField(null=True, default=None)
    options = models.JSONField(default=None, null=True)
    created_by = models.ForeignKey(
        to=SystemUser,
        on_delete=models.CASCADE,
        related_name="answer_history_created",
    )
    created = models.DateTimeField(auto_now_add=True)
    updated = models.DateTimeField(default=None, null=True)

    def __str__(self):
        return self.name

    class Meta:
        db_table = "answer_history"


class ViewPendingDataApproval(models.Model):
    id = models.BigIntegerField(primary_key=True)
    status = models.IntegerField(
        choices=DataApprovalStatus.FieldStr.items(),
        default=DataApprovalStatus.pending,
    )
    user = models.ForeignKey(
        to=SystemUser,
        on_delete=models.DO_NOTHING,
        related_name="user_view_pending_data",
    )
    level = models.ForeignKey(
        to=Levels,
        on_delete=models.DO_NOTHING,
        related_name="level_view_pending_data",
    )
    batch = models.ForeignKey(
        to=PendingDataBatch,
        on_delete=models.DO_NOTHING,
        related_name="batch_view_pending_data",
    )
    pending_level = models.IntegerField()

    class Meta:
        managed = False
        db_table = "view_pending_approval"


class ViewDataOptions(models.Model):
    id = models.BigIntegerField(primary_key=True)
    data = models.ForeignKey(
        to=FormData,
        on_delete=models.DO_NOTHING,
        related_name="data_view_data_options",
    )
    administration = models.ForeignKey(
        to=Administration,
        on_delete=models.PROTECT,
        related_name="administration_view_data_options",
    )
    form = models.ForeignKey(
        to=Forms,
        on_delete=models.DO_NOTHING,
        related_name="form_view_data_options",
    )
    options = models.JSONField(default=None, null=True)

    class Meta:
        managed = False
        db_table = "view_data_options"


class ViewOptions(models.Model):
    id = models.BigIntegerField(primary_key=True)
    data = models.ForeignKey(
        to=FormData,
        on_delete=models.DO_NOTHING,
        related_name="data_view_options",
    )
    administration = models.ForeignKey(
        to=Administration,
        on_delete=models.PROTECT,
        related_name="administration_view_options",
    )
    question = models.ForeignKey(
        to=Questions,
        on_delete=models.DO_NOTHING,
        related_name="question_view_options",
    )
    answer = models.ForeignKey(
        to=Answers,
        on_delete=models.DO_NOTHING,
        related_name="answer_view_options",
    )
    form = models.ForeignKey(
        to=Forms, on_delete=models.DO_NOTHING, related_name="form_view_options"
    )
    options = models.TextField()

    class Meta:
        managed = False
        db_table = "view_options"


class ViewJMPData(models.Model):
    id = models.BigIntegerField(primary_key=True)
    data = models.ForeignKey(
        to=FormData,
        on_delete=models.DO_NOTHING,
        related_name="data_view_jmp_data",
    )
    path = models.TextField(null=True, default=None)
    form = models.ForeignKey(
        to=Forms,
        on_delete=models.DO_NOTHING,
        related_name="form_view_jmp_data",
    )
    name = models.TextField()
    level = models.TextField()
    matches = models.IntegerField()
    score = models.IntegerField()

    @property
    def serialize(self):
        return {
            "data_id": self.data_id,
            "name": self.name,
            "level": self.level,
        }

    class Meta:
        managed = False
        db_table = "view_jmp_data"


class ViewJMPCount(models.Model):
    id = models.BigIntegerField(primary_key=True)
    path = models.TextField(null=True, default=None)
    form = models.ForeignKey(
        to=Forms,
        on_delete=models.DO_NOTHING,
        related_name="form_view_jmp_count",
    )
    name = models.TextField()
    level = models.TextField()
    total = models.IntegerField()

    class Meta:
        managed = False
        db_table = "view_jmp_count"

1	from django.db import models	1✔
2
3	# Create your models here.
4	from api.v1.v1_data.constants import DataApprovalStatus	1✔
5	from api.v1.v1_forms.constants import (	1✔
6	QuestionTypes,
7	SubmissionTypes,
8	)
9	from api.v1.v1_forms.models import Forms, Questions	1✔
10	from api.v1.v1_profile.models import Administration, Levels	1✔
11	from api.v1.v1_users.models import SystemUser	1✔
12	from utils.soft_deletes_model import SoftDeletes	1✔
13	import os	1✔
14	import uuid	1✔
15	import json	1✔
16	from utils import storage	1✔
17
18
19	class FormData(models.Model):	1✔
20	parent = models.ForeignKey(	1✔
21	"self",
22	on_delete=models.CASCADE,
23	related_name="children",
24	null=True,
25	blank=True,
26	)
27	name = models.TextField()	1✔
28	form = models.ForeignKey(	1✔
29	to=Forms, on_delete=models.CASCADE, related_name="form_form_data"
30	)
31	administration = models.ForeignKey(	1✔
32	to=Administration,
33	on_delete=models.PROTECT,
34	related_name="administration_form_data",
35	)
36	geo = models.JSONField(null=True, default=None)	1✔
37	uuid = models.CharField(max_length=255, default=uuid.uuid4, null=True)	1✔
38	submission_type = models.IntegerField(	1✔
39	choices=SubmissionTypes.FieldStr.items(),
40	default=SubmissionTypes.registration,
41	)
42	created_by = models.ForeignKey(	1✔
43	to=SystemUser,
44	on_delete=models.CASCADE,
45	related_name="form_data_created",
46	)
47	updated_by = models.ForeignKey(	1✔
48	to=SystemUser,
49	on_delete=models.CASCADE,
50	related_name="form_data_updated",
51	default=None,
52	null=True,
53	)
54	created = models.DateTimeField(auto_now_add=True)	1✔
55	updated = models.DateTimeField(default=None, null=True)	1✔
56
57	def __str__(self):	1✔
58	return self.name	×
59
60	@property	1✔
61	def to_data_frame(self):	1✔
62	administration = self.administration	1✔
63	data = {	1✔
64	"id": self.id,
65	"datapoint_name": self.name,
66	"administration": administration.administration_column
67	if administration
68	else None,
69	"uuid": self.uuid,
70	"geolocation": f"{self.geo[0]}, {self.geo[1]}"
71	if self.geo
72	else None,
73	"created_by": self.created_by.get_full_name(),
74	"updated_by": self.updated_by.get_full_name()
75	if self.updated_by
76	else None,
77	"created_at": self.created.strftime("%B %d, %Y %I:%M %p"),
78	"updated_at": self.updated.strftime("%B %d, %Y %I:%M %p")
79	if self.updated
80	else None,
81	"submission_type":
82	SubmissionTypes().FieldStr[self.submission_type],
83	}
84	for a in self.data_answer.order_by(	1✔
85	"question__question_group_id", "question__order"
86	).all():
87	data.update(a.to_data_frame)	1✔
88	return data	1✔
89
90	@property	1✔
91	def save_to_file(self):	1✔
92	data = {	1✔
93	"id": self.id,
94	"datapoint_name": self.name,
95	"administration": self.administration.id
96	if self.administration
97	else None,
98	"uuid": str(self.uuid),
99	"geolocation": self.geo,
100	"submission_type": self.submission_type,
101	}
102	answers = {}	1✔
103	for a in self.data_answer.order_by(	1✔
104	"question__question_group_id", "question__order"
105	).all():
106	answers.update(a.to_key)	1✔
107	data.update({"answers": answers})	1✔
108	json_data = json.dumps(data)	1✔
109	file_name = f"{str(self.uuid)}.json"	1✔
110	# write to json file
111	with open(file_name, "w") as f:	1✔
112	f.write(json_data)	1✔
113	storage.upload(file=file_name, folder="datapoints")	1✔
114	# delete file
115	os.remove(file_name)	1✔
116	return data	1✔
117
118	@property	1✔
119	def loc(self):	1✔
120	return self.administration.name	1✔
121
122	class Meta:	1✔
123	db_table = "data"	1✔
124
125
126	class PendingDataBatch(models.Model):	1✔
127	form = models.ForeignKey(	1✔
128	to=Forms, on_delete=models.CASCADE, related_name="form_batch_data"
129	)
130	administration = models.ForeignKey(	1✔
131	to=Administration,
132	on_delete=models.PROTECT,
133	related_name="administration_pending_data_batch",
134	)
135	user = models.ForeignKey(	1✔
136	to=SystemUser,
137	on_delete=models.CASCADE,
138	related_name="user_pending_data_batch",
139	)
140	name = models.TextField()	1✔
141	uuid = models.UUIDField(default=None, null=True)	1✔
142	file = models.URLField(default=None, null=True)	1✔
143	approved = models.BooleanField(default=False)	1✔
144	created = models.DateTimeField(auto_now_add=True)	1✔
145	updated = models.DateTimeField(default=None, null=True)	1✔
146
147	def __str__(self):	1✔
148	return self.name	×
149
150	class Meta:	1✔
151	db_table = "batch"	1✔
152
153
154	class PendingDataBatchComments(models.Model):	1✔
155	batch = models.ForeignKey(	1✔
156	to=PendingDataBatch,
157	on_delete=models.CASCADE,
158	related_name="batch_batch_comment",
159	)
160	user = models.ForeignKey(	1✔
161	to=SystemUser,
162	on_delete=models.CASCADE,
163	related_name="user_batch_comment",
164	)
165	comment = models.TextField()	1✔
166	created = models.DateTimeField(auto_now_add=True)	1✔
167
168	def __str__(self):	1✔
169	return self.comment	×
170
171	class Meta:	1✔
172	db_table = "batch_comment"	1✔
173
174
175	class PendingFormData(SoftDeletes):	1✔
176	name = models.TextField()	1✔
177	form = models.ForeignKey(	1✔
178	to=Forms,
179	on_delete=models.CASCADE,
180	related_name="pending_form_form_data",
181	)
182	data = models.ForeignKey(	1✔
183	to=FormData,
184	on_delete=models.CASCADE,
185	related_name="pending_data_form_data",
186	default=None,
187	null=True,
188	)
189	submission_type = models.IntegerField(	1✔
190	choices=SubmissionTypes.FieldStr.items(),
191	default=SubmissionTypes.registration,
192	)
193	administration = models.ForeignKey(	1✔
194	to=Administration,
195	on_delete=models.PROTECT,
196	related_name="administration_pending_form_data",
197	) # noqa
198	geo = models.JSONField(null=True, default=None)	1✔
199	uuid = models.CharField(max_length=255, default=uuid.uuid4, null=True)	1✔
200	batch = models.ForeignKey(	1✔
201	to=PendingDataBatch,
202	on_delete=models.SET_NULL,
203	default=None,
204	null=True,
205	related_name="batch_pending_data_batch",
206	)
207	created_by = models.ForeignKey(	1✔
208	to=SystemUser,
209	on_delete=models.CASCADE,
210	related_name="pending_form_data_created",
211	)
212	updated_by = models.ForeignKey(	1✔
213	to=SystemUser,
214	on_delete=models.PROTECT,
215	related_name="pending_form_data_updated",
216	default=None,
217	null=True,
218	)
219	created = models.DateTimeField(auto_now_add=True)	1✔
220	updated = models.DateTimeField(default=None, null=True)	1✔
221	duration = models.IntegerField(default=0)	1✔
222	submitter = models.CharField(max_length=255, default=None, null=True)	1✔
223
224	def __str__(self):	1✔
225	return self.name	×
226
227	class Meta:	1✔
228	db_table = "pending_data"	1✔
229
230
231	class PendingDataApproval(models.Model):	1✔
232	batch = models.ForeignKey(	1✔
233	to=PendingDataBatch,
234	on_delete=models.CASCADE,
235	related_name="batch_approval",
236	)
237	user = models.ForeignKey(	1✔
238	to=SystemUser,
239	on_delete=models.CASCADE,
240	related_name="user_assigned_pending_data",
241	)
242	level = models.ForeignKey(	1✔
243	to=Levels,
244	on_delete=models.CASCADE,
245	related_name="level_assigned_pending_data",
246	)
247	status = models.IntegerField(	1✔
248	choices=DataApprovalStatus.FieldStr.items(),
249	default=DataApprovalStatus.pending,
250	)
251
252	def __str__(self):	1✔
253	return self.user.email	×
254
255	class Meta:	1✔
256	db_table = "pending_data_approval"	1✔
257
258
259	class PendingAnswers(models.Model):	1✔
260	pending_data = models.ForeignKey(	1✔
261	to=PendingFormData,
262	on_delete=models.CASCADE,
263	related_name="pending_data_answer",
264	)
265	question = models.ForeignKey(	1✔
266	to=Questions,
267	on_delete=models.CASCADE,
268	related_name="question_pending_answer",
269	)
270	name = models.TextField(null=True, default=None)	1✔
271	value = models.FloatField(null=True, default=None)	1✔
272	options = models.JSONField(default=None, null=True)	1✔
273	created_by = models.ForeignKey(	1✔
274	to=SystemUser,
275	on_delete=models.CASCADE,
276	related_name="pending_answer_created",
277	)
278	created = models.DateTimeField(auto_now_add=True)	1✔
279	updated = models.DateTimeField(default=None, null=True)	1✔
280
281	def __str__(self):	1✔
282	return str(self.id)	×
283
284	class Meta:	1✔
285	db_table = "pending_answer"	1✔
286
287
288	class PendingAnswerHistory(models.Model):	1✔
289	pending_data = models.ForeignKey(	1✔
290	to=PendingFormData,
291	on_delete=models.CASCADE,
292	related_name="pending_data_answer_history",
293	)
294	question = models.ForeignKey(	1✔
295	to=Questions,
296	on_delete=models.CASCADE,
297	related_name="question_pending_answer_history",
298	)
299	name = models.TextField(null=True, default=None)	1✔
300	value = models.FloatField(null=True, default=None)	1✔
301	options = models.JSONField(default=None, null=True)	1✔
302	created_by = models.ForeignKey(	1✔
303	to=SystemUser,
304	on_delete=models.PROTECT,
305	related_name="pending_answer_history_created",
306	)
307	created = models.DateTimeField(auto_now_add=True)	1✔
308	updated = models.DateTimeField(default=None, null=True)	1✔
309
310	def __str__(self):	1✔
311	return self.name	×
312
313	class Meta:	1✔
314	db_table = "pending_answer_history"	1✔
315
316
317	class Answers(models.Model):	1✔
318	data = models.ForeignKey(	1✔
319	to=FormData, on_delete=models.CASCADE, related_name="data_answer"
320	)
321	question = models.ForeignKey(	1✔
322	to=Questions, on_delete=models.CASCADE, related_name="question_answer"
323	)
324	name = models.TextField(null=True, default=None)	1✔
325	value = models.FloatField(null=True, default=None)	1✔
326	options = models.JSONField(default=None, null=True)	1✔
327	created_by = models.ForeignKey(	1✔
328	to=SystemUser, on_delete=models.CASCADE, related_name="answer_created"
329	)
330	created = models.DateTimeField(auto_now_add=True)	1✔
331	updated = models.DateTimeField(default=None, null=True)	1✔
332
333	def __str__(self):	1✔
334	return self.data.name	×
335
336	@property	1✔
337	def to_data_frame(self) -> dict:	1✔
338	q = self.question	1✔
339	qname = f"{self.question.name}"	1✔
340	if q.type in [	1✔
341	QuestionTypes.geo,
342	QuestionTypes.option,
343	QuestionTypes.multiple_option,
344	]:
345	answer = "\|".join(map(str, self.options))	1✔
346	elif q.type in [	1✔
347	QuestionTypes.text,
348	QuestionTypes.photo,
349	QuestionTypes.date,
350	QuestionTypes.autofield,
351	]:
352	answer = self.name	1✔
353	elif q.type == QuestionTypes.administration:	1✔
354	answer = Administration.objects.filter(pk=self.value).first()	1✔
355	if answer:	1!
356	answer = answer.administration_column	1✔
357	else:
358	answer = self.value	1✔
359	return {qname: answer}	1✔
360
361	@property	1✔
362	def to_key(self) -> dict:	1✔
363	q = self.question	1✔
364	if q.type in [	1✔
365	QuestionTypes.geo,
366	QuestionTypes.option,
367	QuestionTypes.multiple_option,
368	]:
369	answer = self.options	1✔
370	elif q.type in [	1✔
371	QuestionTypes.text,
372	QuestionTypes.photo,
373	QuestionTypes.date,
374	QuestionTypes.cascade,
375	]:
376	answer = self.name	1✔
377	else:
378	answer = self.value	1✔
379	return {q.id: answer}	1✔
380
381	class Meta:	1✔
382	db_table = "answer"	1✔
383
384
385	class AnswerHistory(models.Model):	1✔
386	data = models.ForeignKey(	1✔
387	to=FormData,
388	on_delete=models.CASCADE,
389	related_name="data_answer_history",
390	)
391	question = models.ForeignKey(	1✔
392	to=Questions,
393	on_delete=models.CASCADE,
394	related_name="question_answer_history",
395	)
396	name = models.TextField(null=True, default=None)	1✔
397	value = models.FloatField(null=True, default=None)	1✔
398	options = models.JSONField(default=None, null=True)	1✔
399	created_by = models.ForeignKey(	1✔
400	to=SystemUser,
401	on_delete=models.CASCADE,
402	related_name="answer_history_created",
403	)
404	created = models.DateTimeField(auto_now_add=True)	1✔
405	updated = models.DateTimeField(default=None, null=True)	1✔
406
407	def __str__(self):	1✔
408	return self.name	×
409
410	class Meta:	1✔
411	db_table = "answer_history"	1✔
412
413
414	class ViewPendingDataApproval(models.Model):	1✔
415	id = models.BigIntegerField(primary_key=True)	1✔
416	status = models.IntegerField(	1✔
417	choices=DataApprovalStatus.FieldStr.items(),
418	default=DataApprovalStatus.pending,
419	)
420	user = models.ForeignKey(	1✔
421	to=SystemUser,
422	on_delete=models.DO_NOTHING,
423	related_name="user_view_pending_data",
424	)
425	level = models.ForeignKey(	1✔
426	to=Levels,
427	on_delete=models.DO_NOTHING,
428	related_name="level_view_pending_data",
429	)
430	batch = models.ForeignKey(	1✔
431	to=PendingDataBatch,
432	on_delete=models.DO_NOTHING,
433	related_name="batch_view_pending_data",
434	)
435	pending_level = models.IntegerField()	1✔
436
437	class Meta:	1✔
438	managed = False	1✔
439	db_table = "view_pending_approval"	1✔
440
441
442	class ViewDataOptions(models.Model):	1✔
443	id = models.BigIntegerField(primary_key=True)	1✔
444	data = models.ForeignKey(	1✔
445	to=FormData,
446	on_delete=models.DO_NOTHING,
447	related_name="data_view_data_options",
448	)
449	administration = models.ForeignKey(	1✔
450	to=Administration,
451	on_delete=models.PROTECT,
452	related_name="administration_view_data_options",
453	)
454	form = models.ForeignKey(	1✔
455	to=Forms,
456	on_delete=models.DO_NOTHING,
457	related_name="form_view_data_options",
458	)
459	options = models.JSONField(default=None, null=True)	1✔
460
461	class Meta:	1✔
462	managed = False	1✔
463	db_table = "view_data_options"	1✔
464
465
466	class ViewOptions(models.Model):	1✔
467	id = models.BigIntegerField(primary_key=True)	1✔
468	data = models.ForeignKey(	1✔
469	to=FormData,
470	on_delete=models.DO_NOTHING,
471	related_name="data_view_options",
472	)
473	administration = models.ForeignKey(	1✔
474	to=Administration,
475	on_delete=models.PROTECT,
476	related_name="administration_view_options",
477	)
478	question = models.ForeignKey(	1✔
479	to=Questions,
480	on_delete=models.DO_NOTHING,
481	related_name="question_view_options",
482	)
483	answer = models.ForeignKey(	1✔
484	to=Answers,
485	on_delete=models.DO_NOTHING,
486	related_name="answer_view_options",
487	)
488	form = models.ForeignKey(	1✔
489	to=Forms, on_delete=models.DO_NOTHING, related_name="form_view_options"
490	)
491	options = models.TextField()	1✔
492
493	class Meta:	1✔
494	managed = False	1✔
495	db_table = "view_options"	1✔
496
497
498	class ViewJMPData(models.Model):	1✔
499	id = models.BigIntegerField(primary_key=True)	1✔
500	data = models.ForeignKey(	1✔
501	to=FormData,
502	on_delete=models.DO_NOTHING,
503	related_name="data_view_jmp_data",
504	)
505	path = models.TextField(null=True, default=None)	1✔
506	form = models.ForeignKey(	1✔
507	to=Forms,
508	on_delete=models.DO_NOTHING,
509	related_name="form_view_jmp_data",
510	)
511	name = models.TextField()	1✔
512	level = models.TextField()	1✔
513	matches = models.IntegerField()	1✔
514	score = models.IntegerField()	1✔
515
516	@property	1✔
517	def serialize(self):	1✔
518	return {	×
519	"data_id": self.data_id,
520	"name": self.name,
521	"level": self.level,
522	}
523
524	class Meta:	1✔
525	managed = False	1✔
526	db_table = "view_jmp_data"	1✔
527
528
529	class ViewJMPCount(models.Model):	1✔
530	id = models.BigIntegerField(primary_key=True)	1✔
531	path = models.TextField(null=True, default=None)	1✔
532	form = models.ForeignKey(	1✔
533	to=Forms,
534	on_delete=models.DO_NOTHING,
535	related_name="form_view_jmp_count",
536	)
537	name = models.TextField()	1✔
538	level = models.TextField()	1✔
539	total = models.IntegerField()	1✔
540
541	class Meta:	1✔
542	managed = False	1✔
543	db_table = "view_jmp_count"	1✔

akvo / rtmis / #1313

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