akvo / iwsims / #49

Committed 16 Jun 2026 11:40AM UTC coverage: 88.039% (-0.07%) from 88.111%

Build # #49

Build Type

Pull #22

coveralls-python

Committed by

ifirmawan

Commit Message

[#20] Replace all Answers queries with MVAnswerDenormalized across visualization

values_functions.py:
- Replace all 23 Answers.objects queries with MVAnswerDenormalized
  (answer_options GIN index, answer_value numeric index, parent_id direct)
- _number_group_by_parent: 2-query approach (MV for aggregates, FormData
  for parent names) instead of single JOIN — same queries, no Subquery needed
- _option_group_by_option: tracking_field data__parent_id → parent_id (no join)
- Remove Answers import entirely
- Add TruncDate import for _number_group_by_date
- Add MVAnswerDenormalized import

progress_functions.py:
- build_progress_answers_map: MVAnswerDenormalized with field name mapping
  (answer_options/answer_value → options/value for formula handlers)
- handle_progress filter_qid and scope_qid queries: use MVAnswerDenormalized
- Remove Answers import, add MVAnswerDenormalized import

escalation_functions.py:
- build_column_caches: MVAnswerDenormalized for both monitoring and parent
  answer caches with field name normalization
- Remove Answers import entirely

tests/mixins.py:
- Add refresh_all_mvs() at end of VisualizationValuesTestMixin.setUp()
  so all test classes using this mixin automatically have populated MVs
- Extract shared refresh_all_mvs() function and MVRefreshMixin
- Remove MVRefreshMixin from EscalationTestCases (covered by mixin)

Test body fixes (refresh after in-test data creation):
- tests_values_option: test_include_unanswered_works_on_registration_form
- tests_values_criteria: 2 cross-form criteria tests
- tests_values_stack: registration form stack test
- tests_formula_values: setUp override + test_criteria_composes

Shared refresh helpers moved to mixins.py:
- tests_mv_models, tests_mv_helpers, tests_operational_criteria: use
  refresh_all_mvs from mixins instead of local definitions

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>

Pull Request Pull Request #22: [MV4V 4] Optimize Query Functions to Use MVs

Coverage Stats

5087 of 5937 branches covered (85.68%)

Branch coverage included in aggregate %.

9840 of 11018 relevant lines covered (89.31%)

0.89 hits per line

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

69.28

backend/api/v1/v1_visualization/progress_functions.py

from api.v1.v1_data.models import FormData
from api.v1.v1_visualization.functions import (
    apply_administration_filter,
    apply_criteria_to_monitoring_qs,
    apply_parent_criteria_to_qs,
    build_date_filters,
    get_latest_monitoring_subquery,
)
from api.v1.v1_visualization.models import MVAnswerDenormalized


def compute_any_yes(latest_data_id, question_ids, answers_map, **kwargs):
    """100% if ANY listed question answered 'Yes'."""
    for qid in question_ids:
        a = answers_map.get((latest_data_id, qid))
        if a and a.get("options") and "yes" in a["options"]:
            return 100.0
    return 0.0


def compute_completed_binary(
    latest_data_id, question_ids, answers_map, **kwargs
):
    """100% if answered 'Completed'."""
    for qid in question_ids:
        a = answers_map.get((latest_data_id, qid))
        if a and a.get("options") and "completed" in a["options"]:
            return 100.0
    return 0.0


def compute_ratio(
    latest_data_id, question_ids, answers_map, **kwargs
):
    """(Implemented / Planned) * 100, clamped to [0, 100].

    Expects question_ids = [implemented_qid, planned_qid].
    Returns 0.0 if either value is missing or planned <= 0.
    """
    if len(question_ids) < 2:
        return 0.0
    implemented_qid, planned_qid = question_ids[0], question_ids[1]
    impl_row = answers_map.get((latest_data_id, implemented_qid))
    plan_row = answers_map.get((latest_data_id, planned_qid))
    implemented = impl_row.get("value") if impl_row else None
    planned = plan_row.get("value") if plan_row else None
    if implemented is None or planned is None:
        return 0.0
    try:
        planned = float(planned)
        implemented = float(implemented)
    except (TypeError, ValueError):
        return 0.0
    if planned <= 0:
        return 0.0
    return round(min((implemented / planned) * 100, 100.0), 2)


def compute_multi_select_proportion(
    latest_data_id, question_ids, answers_map,
    total_items=1, **kwargs
):
    """Percentage based on number of selected options."""
    selected = None
    for qid in question_ids:
        a = answers_map.get((latest_data_id, qid))
        if a and a.get("options"):
            selected = a["options"]
            break
    if not selected:
        return 0.0
    if not total_items or total_items <= 0:
        return 0.0
    pct = (len(selected) / total_items) * 100
    return round(min(pct, 100.0), 2)


FORMULA_HANDLERS = {
    "any_yes": compute_any_yes,
    "completed_binary": compute_completed_binary,
    "ratio": compute_ratio,
    "multi_select_proportion": (
        compute_multi_select_proportion
    ),
}


def filter_components_by_scope(components, scope_value):
    """Filter components to those applicable for a scope value.

    If scope_value is None or a component has no applicable_types,
    the component is always included.
    """
    if not scope_value:
        return components
    return [
        c for c in components
        if not c.get("applicable_types")
        or scope_value in c["applicable_types"]
    ]


def compute_component_scores(
    latest_id, components, answers_map, scope_value=None,
):
    """Compute progress scores for applicable components."""
    active = filter_components_by_scope(components, scope_value)
    scores = {}
    for comp in active:
        handler = FORMULA_HANDLERS[comp["formula"]]
        kwargs = {}
        if comp.get("total_items"):
            kwargs["total_items"] = comp["total_items"]
        scores[comp["key"]] = handler(
            latest_id,
            comp["question_ids"],
            answers_map,
            **kwargs,
        )
    return scores


def build_progress_answers_map(latest_ids, components):
    """Bulk-fetch answers needed to score all components for all parents.

    Returns dict keyed by (data_id, question_id) carrying the fields
    formula handlers read (options, value).
    """
    qids = {
        q for c in components for q in c.get("question_ids", [])
    }
    if not qids or not latest_ids:
        return {}
    rows = MVAnswerDenormalized.objects.filter(
        data_id__in=latest_ids,
        question_id__in=qids,
    ).values(
        "data_id", "question_id",
        "answer_options", "answer_value",
    )
    return {
        (r["data_id"], r["question_id"]): {
            "options": r["answer_options"],
            "value": r["answer_value"],
        }
        for r in rows
    }


def build_histogram(eps_results):
    """Bucket overall progress into 10% ranges.

    Returns list of 10 buckets, each with progress label
    and count.
    """
    buckets = [
        "0-10%", "11-20%", "21-30%", "31-40%", "41-50%",
        "51-60%", "61-70%", "71-80%", "81-90%", "91-100%",
    ]
    counts = [0] * 10
    for eps in eps_results:
        overall = eps["overall"]
        if overall <= 0:
            idx = 0
        else:
            idx = min(max(0, int(overall - 1) // 10), 9)
        counts[idx] += 1
    return [
        {"progress": buckets[i], "count": counts[i]}
        for i in range(10)
    ]


def handle_progress(
    parent_form, monitoring_form_id,
    components, params,
):
    """Handle progress query.

    Args:
        parent_form: Registration form instance.
        monitoring_form_id: Monitoring form ID.
        components: Parsed component list with
            key/formula/question_ids/total_items.
        params: Dict with filter_question_id,
            filter_option_value, administration_id,
            from_date, to_date, date_question_id.

    Returns:
        Dict with histogram and details.
    """
    administration_id = params.get("administration_id")
    filter_qid = params.get("filter_question_id")
    filter_value = params.get("filter_option_value")

    date_filters = build_date_filters(params)

    parents = FormData.objects.filter(
        form=parent_form,
        parent__isnull=True,
        is_pending=False,
        is_draft=False,
    ).annotate(
        latest_id=get_latest_monitoring_subquery(
            monitoring_form_id, date_filters or None
        ),
    ).filter(latest_id__isnull=False)

    if administration_id:
        parents = apply_administration_filter(
            parents, administration_id
        )

    # Multi-criteria AND filter (shared grammar with /values).
    parents = apply_criteria_to_monitoring_qs(
        parents, True, params.get("criteria"),
    )
    parents = apply_parent_criteria_to_qs(
        parents, True, params.get("parent_criteria"),
    )

    # Optional filter by latest monitoring option value
    if filter_qid and filter_value:
        latest_ids = parents.values_list(
            "latest_id", flat=True
        )
        matching_ids = MVAnswerDenormalized.objects.filter(
            data_id__in=latest_ids,
            question_id=filter_qid,
            answer_options__contains=[filter_value],
        ).values_list("data_id", flat=True)
        parents = parents.filter(
            latest_id__in=matching_ids
        )

    # Compute scores per parent
    scope_qid = params.get("scope_question_id")
    parents = list(parents.only("id", "name"))
    latest_ids = [p.latest_id for p in parents]
    answers_map = build_progress_answers_map(
        latest_ids, components
    )

    # Build scope lookup: latest_id -> scope option value
    scope_map = {}
    if scope_qid:
        scope_rows = MVAnswerDenormalized.objects.filter(
            data_id__in=latest_ids,
            question_id=scope_qid,
        ).values("data_id", "answer_options")
        for row in scope_rows:
            opts = row.get("answer_options") or []
            if opts:
                scope_map[row["data_id"]] = opts[0]

    eps_results = []
    for parent in parents:
        scope_value = scope_map.get(parent.latest_id)
        scores = compute_component_scores(
            parent.latest_id, components, answers_map,
            scope_value=scope_value,
        )
        overall = (
            round(sum(scores.values()) / len(scores), 2)
            if scores else 0.0
        )
        eps_results.append({
            "label": parent.name,
            "group": str(parent.id),
            "components": scores,
            "overall": overall,
        })

    histogram = build_histogram(eps_results)

    return {
        "histogram": histogram,
        "details": eps_results,
    }

1	from api.v1.v1_data.models import FormData	1✔
2	from api.v1.v1_visualization.functions import (	1✔
3	apply_administration_filter,
4	apply_criteria_to_monitoring_qs,
5	apply_parent_criteria_to_qs,
6	build_date_filters,
7	get_latest_monitoring_subquery,
8	)
9	from api.v1.v1_visualization.models import MVAnswerDenormalized	1✔
10
11
12	def compute_any_yes(latest_data_id, question_ids, answers_map, **kwargs):	1✔
13	"""100% if ANY listed question answered 'Yes'."""
14	for qid in question_ids:	×
15	a = answers_map.get((latest_data_id, qid))	×
16	if a and a.get("options") and "yes" in a["options"]:	×
17	return 100.0	×
18	return 0.0	×
19
20
21	def compute_completed_binary(	1✔
22	latest_data_id, question_ids, answers_map, **kwargs
23	):
24	"""100% if answered 'Completed'."""
25	for qid in question_ids:	×
26	a = answers_map.get((latest_data_id, qid))	×
27	if a and a.get("options") and "completed" in a["options"]:	×
28	return 100.0	×
29	return 0.0	×
30
31
32	def compute_ratio(	1✔
33	latest_data_id, question_ids, answers_map, **kwargs
34	):
35	"""(Implemented / Planned) * 100, clamped to [0, 100].
36
37	Expects question_ids = [implemented_qid, planned_qid].
38	Returns 0.0 if either value is missing or planned <= 0.
39	"""
40	if len(question_ids) < 2:	1!
41	return 0.0	×
42	implemented_qid, planned_qid = question_ids[0], question_ids[1]	1✔
43	impl_row = answers_map.get((latest_data_id, implemented_qid))	1✔
44	plan_row = answers_map.get((latest_data_id, planned_qid))	1✔
45	implemented = impl_row.get("value") if impl_row else None	1✔
46	planned = plan_row.get("value") if plan_row else None	1✔
47	if implemented is None or planned is None:	1!
48	return 0.0	×
49	try:	1✔
50	planned = float(planned)	1✔
51	implemented = float(implemented)	1✔
52	except (TypeError, ValueError):	×
53	return 0.0	×
54	if planned <= 0:	1!
55	return 0.0	×
56	return round(min((implemented / planned) * 100, 100.0), 2)	1✔
57
58
59	def compute_multi_select_proportion(	1✔
60	latest_data_id, question_ids, answers_map,
61	total_items=1, **kwargs
62	):
63	"""Percentage based on number of selected options."""
64	selected = None	1✔
65	for qid in question_ids:	1!
66	a = answers_map.get((latest_data_id, qid))	1✔
67	if a and a.get("options"):	1!
68	selected = a["options"]	1✔
69	break	1✔
70	if not selected:	1!
71	return 0.0	×
72	if not total_items or total_items <= 0:	1!
73	return 0.0	×
74	pct = (len(selected) / total_items) * 100	1✔
75	return round(min(pct, 100.0), 2)	1✔
76
77
78	FORMULA_HANDLERS = {	1✔
79	"any_yes": compute_any_yes,
80	"completed_binary": compute_completed_binary,
81	"ratio": compute_ratio,
82	"multi_select_proportion": (
83	compute_multi_select_proportion
84	),
85	}
86
87
88	def filter_components_by_scope(components, scope_value):	1✔
89	"""Filter components to those applicable for a scope value.
90
91	If scope_value is None or a component has no applicable_types,
92	the component is always included.
93	"""
94	if not scope_value:	1!
95	return components	1✔
96	return [	×
97	c for c in components
98	if not c.get("applicable_types")
99	or scope_value in c["applicable_types"]
100	]
101
102
103	def compute_component_scores(	1✔
104	latest_id, components, answers_map, scope_value=None,
105	):
106	"""Compute progress scores for applicable components."""
107	active = filter_components_by_scope(components, scope_value)	1✔
108	scores = {}	1✔
109	for comp in active:	1✔
110	handler = FORMULA_HANDLERS[comp["formula"]]	1✔
111	kwargs = {}	1✔
112	if comp.get("total_items"):	1✔
113	kwargs["total_items"] = comp["total_items"]	1✔
114	scores[comp["key"]] = handler(	1✔
115	latest_id,
116	comp["question_ids"],
117	answers_map,
118	**kwargs,
119	)
120	return scores	1✔
121
122
123	def build_progress_answers_map(latest_ids, components):	1✔
124	"""Bulk-fetch answers needed to score all components for all parents.
125
126	Returns dict keyed by (data_id, question_id) carrying the fields
127	formula handlers read (options, value).
128	"""
129	qids = {	1✔
130	q for c in components for q in c.get("question_ids", [])
131	}
132	if not qids or not latest_ids:	1!
133	return {}	×
134	rows = MVAnswerDenormalized.objects.filter(	1✔
135	data_id__in=latest_ids,
136	question_id__in=qids,
137	).values(
138	"data_id", "question_id",
139	"answer_options", "answer_value",
140	)
141	return {	1✔
142	(r["data_id"], r["question_id"]): {
143	"options": r["answer_options"],
144	"value": r["answer_value"],
145	}
146	for r in rows
147	}
148
149
150	def build_histogram(eps_results):	1✔
151	"""Bucket overall progress into 10% ranges.
152
153	Returns list of 10 buckets, each with progress label
154	and count.
155	"""
156	buckets = [	1✔
157	"0-10%", "11-20%", "21-30%", "31-40%", "41-50%",
158	"51-60%", "61-70%", "71-80%", "81-90%", "91-100%",
159	]
160	counts = [0] * 10	1✔
161	for eps in eps_results:	1✔
162	overall = eps["overall"]	1✔
163	if overall <= 0:	1!
164	idx = 0	×
165	else:
166	idx = min(max(0, int(overall - 1) // 10), 9)	1✔
167	counts[idx] += 1	1✔
168	return [	1✔
169	{"progress": buckets[i], "count": counts[i]}
170	for i in range(10)
171	]
172
173
174	def handle_progress(	1✔
175	parent_form, monitoring_form_id,
176	components, params,
177	):
178	"""Handle progress query.
179
180	Args:
181	parent_form: Registration form instance.
182	monitoring_form_id: Monitoring form ID.
183	components: Parsed component list with
184	key/formula/question_ids/total_items.
185	params: Dict with filter_question_id,
186	filter_option_value, administration_id,
187	from_date, to_date, date_question_id.
188
189	Returns:
190	Dict with histogram and details.
191	"""
192	administration_id = params.get("administration_id")	1✔
193	filter_qid = params.get("filter_question_id")	1✔
194	filter_value = params.get("filter_option_value")	1✔
195
196	date_filters = build_date_filters(params)	1✔
197
198	parents = FormData.objects.filter(	1✔
199	form=parent_form,
200	parent__isnull=True,
201	is_pending=False,
202	is_draft=False,
203	).annotate(
204	latest_id=get_latest_monitoring_subquery(
205	monitoring_form_id, date_filters or None
206	),
207	).filter(latest_id__isnull=False)
208
209	if administration_id:	1!
210	parents = apply_administration_filter(	1✔
211	parents, administration_id
212	)
213
214	# Multi-criteria AND filter (shared grammar with /values).
215	parents = apply_criteria_to_monitoring_qs(	1✔
216	parents, True, params.get("criteria"),
217	)
218	parents = apply_parent_criteria_to_qs(	1✔
219	parents, True, params.get("parent_criteria"),
220	)
221
222	# Optional filter by latest monitoring option value
223	if filter_qid and filter_value:	1✔
224	latest_ids = parents.values_list(	1✔
225	"latest_id", flat=True
226	)
227	matching_ids = MVAnswerDenormalized.objects.filter(	1✔
228	data_id__in=latest_ids,
229	question_id=filter_qid,
230	answer_options__contains=[filter_value],
231	).values_list("data_id", flat=True)
232	parents = parents.filter(	1✔
233	latest_id__in=matching_ids
234	)
235
236	# Compute scores per parent
237	scope_qid = params.get("scope_question_id")	1✔
238	parents = list(parents.only("id", "name"))	1✔
239	latest_ids = [p.latest_id for p in parents]	1✔
240	answers_map = build_progress_answers_map(	1✔
241	latest_ids, components
242	)
243
244	# Build scope lookup: latest_id -> scope option value
245	scope_map = {}	1✔
246	if scope_qid:	1!
247	scope_rows = MVAnswerDenormalized.objects.filter(	×
248	data_id__in=latest_ids,
249	question_id=scope_qid,
250	).values("data_id", "answer_options")
251	for row in scope_rows:	×
252	opts = row.get("answer_options") or []	×
253	if opts:	×
254	scope_map[row["data_id"]] = opts[0]	×
255
256	eps_results = []	1✔
257	for parent in parents:	1✔
258	scope_value = scope_map.get(parent.latest_id)	1✔
259	scores = compute_component_scores(	1✔
260	parent.latest_id, components, answers_map,
261	scope_value=scope_value,
262	)
263	overall = (	1✔
264	round(sum(scores.values()) / len(scores), 2)
265	if scores else 0.0
266	)
267	eps_results.append({	1✔
268	"label": parent.name,
269	"group": str(parent.id),
270	"components": scores,
271	"overall": overall,
272	})
273
274	histogram = build_histogram(eps_results)	1✔
275
276	return {	1✔
277	"histogram": histogram,
278	"details": eps_results,
279	}

akvo / iwsims / #49

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