1346

pending completion

Build # 1346

Build Type

Pull #656

travis-ci

Committed by

web-flow

Commit Message

Ignore W503 lint error

Let's follow the directions of Donald E. Knuth :)

"Although formulas within a paragraph always break after binary
operations and relations, displayed formulas always break before binary
operations"

Reference
---
1. https://www.flake8rules.com/rules/W503.html
2. https://www.python.org/dev/peps/pep-0008/#should-a-line-break-before-or-after-a-binary-operator
3. https://github.com/python/peps/commit/c59c4376a

Pull Request Pull Request #656: Style the code consistently

Run Details

492 of 492 new or added lines in 40 files covered. (100.0%)

1363 of 2026 relevant lines covered (67.28%)

2.02 hits per line

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

76.07

/junction/feedback/service.py

# -*- coding: utf-8 -*-

# Standard Library
from collections import defaultdict

# Third Party Stuff
from django.core.exceptions import ObjectDoesNotExist
from django.db import IntegrityError, transaction
from django.db.models import Count

# Junction Stuff
from junction.devices.models import Device
from junction.schedule.models import ScheduleItem, ScheduleItemType

from .models import (
    ChoiceFeedbackQuestion,
    ChoiceFeedbackQuestionValue,
    ScheduleItemChoiceFeedback,
    ScheduleItemTextFeedback,
    TextFeedbackQuestion,
)

COLORS = ["#46BFBD", "#FDB45C", "#F7464A"]


def get_feedback_questions(conference_id):
    """Get all feedback questions for the conference.

    {'talk': {'Text': [{'id': 1, 'title': 'How was the speaker ?',
    'is_required': True}], 'Workshop': [{'id': 1,
    'title': 'How was the content ?', 'is_required': True,
    allowed_values: [{'title': 'Awesome', 'id': 2},
    {'title': 'Bad', 'id': 3}, {'title': 'Ok', 'id': 4}}]}]
    }}
    """
    text_questions = get_text_feedback_questions(conference_id=conference_id)
    choice_questions = get_choice_feedback_questions(conference_id=conference_id)
    return _merge_questions(
        text_questions=text_questions, choice_questions=choice_questions
    )


def get_text_feedback_questions(conference_id):
    """Get all text questions for the conference organized by
    schedule item type.

    Return dict contain all questions with schedule item type in dict.
    """
    qs = TextFeedbackQuestion.objects.filter(conference_id=conference_id)
    return _get_question_oragnized_by_type(qs)


def get_choice_feedback_questions(conference_id):
    """Get all choice based questions for the conference organized by
    schedule item type.
    """
    qs = ChoiceFeedbackQuestion.objects.filter(
        conference_id=conference_id
    ).prefetch_related("allowed_values")
    return _get_question_oragnized_by_type(qs)


def has_submitted(feedback, device_uuid):
    """
    """
    device = Device.objects.get(uuid=device_uuid)
    text_feedback = ScheduleItemTextFeedback.objects.filter(
        schedule_item_id=feedback.validated_data["schedule_item_id"], device=device
    )
    if text_feedback:
        return True
    choice_feedback = ScheduleItemChoiceFeedback.objects.filter(
        schedule_item_id=feedback.validated_data["schedule_item_id"], device=device
    )
    return choice_feedback


def _has_required_ids(master, submitted):
    for item in master:
        if item not in submitted:
            return False
    return True


def has_required_fields_data(feedback):
    try:
        data = feedback.validated_data
        sch = ScheduleItem.objects.get(pk=data["schedule_item_id"])
        sch_type = ScheduleItemType.objects.get(title=sch.type)

        t_ids = TextFeedbackQuestion.objects.filter(
            schedule_item_type=sch_type, conference=sch.conference, is_required=True
        ).values_list("id", flat=True)

        if not data.get("text"):
            if t_ids:
                return False, "Text Feedback is missing"
        else:
            submitted_t_ids = {d["id"] for d in data.get("text")}

            if not _has_required_ids(master=t_ids, submitted=submitted_t_ids):
                return False, "Required text questions are missing"

        c_ids = ChoiceFeedbackQuestion.objects.filter(
            schedule_item_type=sch_type, conference=sch.conference, is_required=True
        ).values_list("id", flat=True)

        if not data.get("choices"):
            if c_ids:
                return False, "Choice feedback is missing"
        else:
            submitted_c_ids = {d["id"] for d in data.get("choices")}

            if not _has_required_ids(master=c_ids, submitted=submitted_c_ids):
                return False, "Choice feedback is missing"
        return True, ""
    except ObjectDoesNotExist as e:
        print(e)
        return False


def create(feedback, device_uuid):
    device = Device.objects.get(uuid=device_uuid)
    schedule_item_id = feedback.validated_data["schedule_item_id"]
    try:
        with transaction.atomic():
            text, choices = [], []
            if feedback.validated_data.get("text"):
                text = create_text_feedback(
                    schedule_item_id=schedule_item_id,
                    feedbacks=feedback.validated_data.get("text"),
                    device=device,
                )
            if feedback.validated_data.get("choices"):
                choices = create_choice_feedback(
                    schedule_item_id=schedule_item_id,
                    feedbacks=feedback.validated_data.get("choices"),
                    device=device,
                )
            return {"text": text, "choices": choices}
    except (IntegrityError, ObjectDoesNotExist) as e:
        print(e)  # Replace with log
        return False


def create_text_feedback(schedule_item_id, feedbacks, device):
    text = []
    for feedback in feedbacks:
        obj = ScheduleItemTextFeedback.objects.create(
            schedule_item_id=schedule_item_id,
            question_id=feedback["id"],
            text=feedback["text"],
            device=device,
        )
        d = {
            "id": obj.id,
            "text": obj.text,
            "question_id": feedback["id"],
            "schedule_item_id": schedule_item_id,
        }
        text.append(d)
    return text


def create_choice_feedback(schedule_item_id, feedbacks, device):
    choices = []
    for feedback in feedbacks:
        value = ChoiceFeedbackQuestionValue.objects.get(
            question_id=feedback["id"], id=feedback["value_id"]
        )
        obj = ScheduleItemChoiceFeedback.objects.create(
            schedule_item_id=schedule_item_id,
            device=device,
            question_id=feedback["id"],
            value=value.value,
        )
        d = {
            "id": obj.id,
            "value_id": value.id,
            "question_id": feedback["id"],
            "schedule_item_id": schedule_item_id,
        }
        choices.append(d)
    return choices


def get_feedback(schedule_item):
    feedback = {
        "text": _get_text_feedback(schedule_item=schedule_item),
        "choices": _get_choice_feedback(schedule_item=schedule_item),
    }
    return feedback


def _get_text_feedback(schedule_item):
    questions = TextFeedbackQuestion.objects.filter(
        schedule_item_type__title=schedule_item.type
    )
    text = [
        {
            "question": question,
            "values": ScheduleItemTextFeedback.objects.filter(
                question=question, schedule_item=schedule_item
            ),
        }
        for question in questions
    ]
    return text


def _get_choice_feedback(schedule_item):
    questions = ChoiceFeedbackQuestion.objects.filter(
        schedule_item_type__title=schedule_item.type
    ).select_related("allowed_values")
    choices = []
    for question in questions:
        values = (
            ScheduleItemChoiceFeedback.objects.filter(
                schedule_item=schedule_item, question=question
            )
            .values("value")
            .annotate(Count("value"))
        )
        d = {
            "question": question,
            "values": _get_choice_value_for_chart(question=question, values=values),
        }
        choices.append(d)
    return choices


def _get_choice_value_for_chart(question, values):
    data = []
    for index, value in enumerate(values):
        d = {"label": str(question.allowed_values.get(value=value["value"]).title)}
        d["value"] = value["value__count"]
        d["color"] = COLORS[index]
        data.append(d)
    return data


def _get_question_oragnized_by_type(qs):
    questions = defaultdict(list)
    for question in qs:
        questions[question.schedule_item_type.title].append(question.to_response())
    return questions


def _merge_questions(text_questions, choice_questions):
    """Merge the choice and text based questions into schedule type
    {'Talk': {'text': [..], 'choice': [...]},}
    """
    types = set(text_questions.keys())
    types.union(list(choice_questions.keys()))
    questions = {}
    for item in types:
        questions[item] = {
            "text": text_questions.get(item),
            "choice": choice_questions.get(item),
        }
    return questions

1	# -- coding: utf-8 --
2
3	# Standard Library
4	from collections import defaultdict	3✔
5
6	# Third Party Stuff
7	from django.core.exceptions import ObjectDoesNotExist	3✔
8	from django.db import IntegrityError, transaction	3✔
9	from django.db.models import Count	3✔
10
11	# Junction Stuff
12	from junction.devices.models import Device	3✔
13	from junction.schedule.models import ScheduleItem, ScheduleItemType	3✔
14
15	from .models import (	3✔
16	ChoiceFeedbackQuestion,
17	ChoiceFeedbackQuestionValue,
18	ScheduleItemChoiceFeedback,
19	ScheduleItemTextFeedback,
20	TextFeedbackQuestion,
21	)
22
23	COLORS = ["#46BFBD", "#FDB45C", "#F7464A"]	3✔
24
25
26	def get_feedback_questions(conference_id):	3✔
27	"""Get all feedback questions for the conference.
28
29	{'talk': {'Text': [{'id': 1, 'title': 'How was the speaker ?',
30	'is_required': True}], 'Workshop': [{'id': 1,
31	'title': 'How was the content ?', 'is_required': True,
32	allowed_values: [{'title': 'Awesome', 'id': 2},
33	{'title': 'Bad', 'id': 3}, {'title': 'Ok', 'id': 4}}]}]
34	}}
35	"""
36	text_questions = get_text_feedback_questions(conference_id=conference_id)	3✔
37	choice_questions = get_choice_feedback_questions(conference_id=conference_id)	3✔
38	return _merge_questions(	3✔
39	text_questions=text_questions, choice_questions=choice_questions
40	)
41
42
43	def get_text_feedback_questions(conference_id):	3✔
44	"""Get all text questions for the conference organized by
45	schedule item type.
46
47	Return dict contain all questions with schedule item type in dict.
48	"""
49	qs = TextFeedbackQuestion.objects.filter(conference_id=conference_id)	3✔
50	return _get_question_oragnized_by_type(qs)	3✔
51
52
53	def get_choice_feedback_questions(conference_id):	3✔
54	"""Get all choice based questions for the conference organized by
55	schedule item type.
56	"""
57	qs = ChoiceFeedbackQuestion.objects.filter(	3✔
58	conference_id=conference_id
59	).prefetch_related("allowed_values")
60	return _get_question_oragnized_by_type(qs)	3✔
61
62
63	def has_submitted(feedback, device_uuid):	3✔
64	"""
65	"""
66	device = Device.objects.get(uuid=device_uuid)	3✔
67	text_feedback = ScheduleItemTextFeedback.objects.filter(	3✔
68	schedule_item_id=feedback.validated_data["schedule_item_id"], device=device
69	)
70	if text_feedback:	3✔
71	return True	3✔
72	choice_feedback = ScheduleItemChoiceFeedback.objects.filter(	3✔
73	schedule_item_id=feedback.validated_data["schedule_item_id"], device=device
74	)
75	return choice_feedback	3✔
76
77
78	def _has_required_ids(master, submitted):	3✔
79	for item in master:	3✔
80	if item not in submitted:	3✔
81	return False	×
82	return True	3✔
83
84
85	def has_required_fields_data(feedback):	3✔
86	try:	3✔
87	data = feedback.validated_data	3✔
88	sch = ScheduleItem.objects.get(pk=data["schedule_item_id"])	3✔
89	sch_type = ScheduleItemType.objects.get(title=sch.type)	3✔
90
91	t_ids = TextFeedbackQuestion.objects.filter(	3✔
92	schedule_item_type=sch_type, conference=sch.conference, is_required=True
93	).values_list("id", flat=True)
94
95	if not data.get("text"):	3✔
96	if t_ids:	3✔
97	return False, "Text Feedback is missing"	3✔
98	else:
99	submitted_t_ids = {d["id"] for d in data.get("text")}	3✔
100
101	if not _has_required_ids(master=t_ids, submitted=submitted_t_ids):	3✔
102	return False, "Required text questions are missing"	×
103
104	c_ids = ChoiceFeedbackQuestion.objects.filter(	3✔
105	schedule_item_type=sch_type, conference=sch.conference, is_required=True
106	).values_list("id", flat=True)
107
108	if not data.get("choices"):	3✔
109	if c_ids:	3✔
110	return False, "Choice feedback is missing"	3✔
111	else:
112	submitted_c_ids = {d["id"] for d in data.get("choices")}	3✔
113
114	if not _has_required_ids(master=c_ids, submitted=submitted_c_ids):	3✔
115	return False, "Choice feedback is missing"	×
116	return True, ""	3✔
117	except ObjectDoesNotExist as e:	×
118	print(e)	×
119	return False	×
120
121
122	def create(feedback, device_uuid):	3✔
123	device = Device.objects.get(uuid=device_uuid)	3✔
124	schedule_item_id = feedback.validated_data["schedule_item_id"]	3✔
125	try:	3✔
126	with transaction.atomic():	3✔
127	text, choices = [], []	3✔
128	if feedback.validated_data.get("text"):	3✔
129	text = create_text_feedback(	3✔
130	schedule_item_id=schedule_item_id,
131	feedbacks=feedback.validated_data.get("text"),
132	device=device,
133	)
134	if feedback.validated_data.get("choices"):	3✔
135	choices = create_choice_feedback(	3✔
136	schedule_item_id=schedule_item_id,
137	feedbacks=feedback.validated_data.get("choices"),
138	device=device,
139	)
140	return {"text": text, "choices": choices}	3✔
141	except (IntegrityError, ObjectDoesNotExist) as e:	×
142	print(e) # Replace with log	×
143	return False	×
144
145
146	def create_text_feedback(schedule_item_id, feedbacks, device):	3✔
147	text = []	3✔
148	for feedback in feedbacks:	3✔
149	obj = ScheduleItemTextFeedback.objects.create(	3✔
150	schedule_item_id=schedule_item_id,
151	question_id=feedback["id"],
152	text=feedback["text"],
153	device=device,
154	)
155	d = {	3✔
156	"id": obj.id,
157	"text": obj.text,
158	"question_id": feedback["id"],
159	"schedule_item_id": schedule_item_id,
160	}
161	text.append(d)	3✔
162	return text	3✔
163
164
165	def create_choice_feedback(schedule_item_id, feedbacks, device):	3✔
166	choices = []	3✔
167	for feedback in feedbacks:	3✔
168	value = ChoiceFeedbackQuestionValue.objects.get(	3✔
169	question_id=feedback["id"], id=feedback["value_id"]
170	)
171	obj = ScheduleItemChoiceFeedback.objects.create(	3✔
172	schedule_item_id=schedule_item_id,
173	device=device,
174	question_id=feedback["id"],
175	value=value.value,
176	)
177	d = {	3✔
178	"id": obj.id,
179	"value_id": value.id,
180	"question_id": feedback["id"],
181	"schedule_item_id": schedule_item_id,
182	}
183	choices.append(d)	3✔
184	return choices	3✔
185
186
187	def get_feedback(schedule_item):	3✔
188	feedback = {	×
189	"text": _get_text_feedback(schedule_item=schedule_item),
190	"choices": _get_choice_feedback(schedule_item=schedule_item),
191	}
192	return feedback	×
193
194
195	def _get_text_feedback(schedule_item):	3✔
196	questions = TextFeedbackQuestion.objects.filter(	×
197	schedule_item_type__title=schedule_item.type
198	)
199	text = [	×
200	{
201	"question": question,
202	"values": ScheduleItemTextFeedback.objects.filter(
203	question=question, schedule_item=schedule_item
204	),
205	}
206	for question in questions
207	]
208	return text	×
209
210
211	def _get_choice_feedback(schedule_item):	3✔
212	questions = ChoiceFeedbackQuestion.objects.filter(	×
213	schedule_item_type__title=schedule_item.type
214	).select_related("allowed_values")
215	choices = []	×
216	for question in questions:	×
217	values = (	×
218	ScheduleItemChoiceFeedback.objects.filter(
219	schedule_item=schedule_item, question=question
220	)
221	.values("value")
222	.annotate(Count("value"))
223	)
224	d = {	×
225	"question": question,
226	"values": _get_choice_value_for_chart(question=question, values=values),
227	}
228	choices.append(d)	×
229	return choices	×
230
231
232	def _get_choice_value_for_chart(question, values):	3✔
233	data = []	×
234	for index, value in enumerate(values):	×
235	d = {"label": str(question.allowed_values.get(value=value["value"]).title)}	×
236	d["value"] = value["value__count"]	×
237	d["color"] = COLORS[index]	×
238	data.append(d)	×
239	return data	×
240
241
242	def _get_question_oragnized_by_type(qs):	3✔
243	questions = defaultdict(list)	3✔
244	for question in qs:	3✔
245	questions[question.schedule_item_type.title].append(question.to_response())	3✔
246	return questions	3✔
247
248
249	def _merge_questions(text_questions, choice_questions):	3✔
250	"""Merge the choice and text based questions into schedule type
251	{'Talk': {'text': [..], 'choice': [...]},}
252	"""
253	types = set(text_questions.keys())	3✔
254	types.union(list(choice_questions.keys()))	3✔
255	questions = {}	3✔
256	for item in types:	3✔
257	questions[item] = {	3✔
258	"text": text_questions.get(item),
259	"choice": choice_questions.get(item),
260	}
261	return questions	3✔

pythonindia / junction / 1346

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