Coverage for rest_api/viewsets_statistics_and_plots.py: 20%
238 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-10 03:22 +0000
1import hashlib
3from django.core.cache import cache
4from django.db.models import BooleanField
5from django.db.models import Case
6from django.db.models import CharField
7from django.db.models import Count
8from django.db.models import Exists
9from django.db.models import OuterRef
10from django.db.models import Q
11from django.db.models import Value
12from django.db.models import When
13from django.db.models.functions import TruncWeek
14from rest_framework import generics
15from rest_framework import status
16from rest_framework import viewsets
17from rest_framework.decorators import action
18from rest_framework.request import Request
19from rest_framework.response import Response
21from rest_api.viewsets import PropertyViewSet
22from rest_api.viewsets_sample import SampleFilterMixin
23from sonar_backend.settings import CACHE_OBJECT_TTL
24from sonar_backend.settings import LOGGER
25from . import models
28class SampleViewSetStatistics(
29 SampleFilterMixin,
30 viewsets.GenericViewSet,
31 generics.mixins.ListModelMixin,
32 generics.mixins.RetrieveModelMixin,
33):
34 @staticmethod
35 def get_statistics(reference=None):
36 response_dict = {}
37 queryset = models.Sample.objects.all()
38 if reference:
39 queryset = queryset.filter(
40 sequences__alignments__replicon__reference__accession=reference
41 ).distinct()
42 response_dict["samples_total"] = queryset.count()
44 first_sample = (
45 queryset.filter(collection_date__isnull=False)
46 .order_by("collection_date")
47 .first()
48 )
49 response_dict["first_sample_date"] = (
50 first_sample.collection_date if first_sample else None
51 )
53 latest_sample = (
54 queryset.filter(collection_date__isnull=False)
55 .order_by("-collection_date")
56 .first()
57 )
58 response_dict["latest_sample_date"] = (
59 latest_sample.collection_date if latest_sample else None
60 )
61 response_dict["populated_metadata_fields"] = (
62 SampleViewSetStatistics().get_populated_metadata_fields(queryset=queryset)
63 )
65 return response_dict
67 @staticmethod
68 def get_populated_metadata_fields(queryset):
69 queryset = queryset.prefetch_related("properties__property")
70 annotations = {}
72 # check profiles was removed because it is slow and because these fields
73 # are basically always populated
75 # check sample fields
76 for field in models.Sample._meta.get_fields():
77 if field.concrete and not field.is_relation:
78 field_name = field.name
79 if field.get_internal_type() == "CharField":
80 condition = Q(**{f"{field_name}__isnull": False}) & ~Q(
81 **{field_name: ""}
82 )
83 else:
84 condition = Q(**{f"{field_name}__isnull": False})
86 annotations[f"has_{field_name}"] = Case(
87 When(condition, then=True),
88 default=False,
89 output_field=BooleanField(),
90 )
92 # check properties
93 property_to_datatype = dict(
94 models.Property.objects.values_list("name", "datatype")
95 )
96 for property_name in PropertyViewSet.get_custom_property_names():
97 datatype = property_to_datatype.get(property_name)
98 if not datatype:
99 LOGGER.warning(f"Property {property_name} not found")
100 continue
102 try:
103 annotations[f"has_{property_name}"] = Exists(
104 models.Sample.objects.filter(
105 id=OuterRef("id"),
106 properties__property__name=property_name,
107 **{f"properties__{datatype}__isnull": False},
108 )
109 )
110 except Exception as e:
111 LOGGER.error(
112 f"Error processing property {property_name} (datatype: {datatype}): {e}"
113 )
115 # apply annotations and check existence (True/False)
116 result = queryset.annotate(**annotations).values(*annotations.keys()).first()
118 return (
119 [k.replace("has_", "") for k in filter(result.get, result)]
120 if result
121 else []
122 )
124 @action(detail=False, methods=["get"])
125 def statistics(self, request: Request, *args, **kwargs):
126 response_dict = self.get_statistics(
127 reference=request.query_params.get("reference")
128 )
129 return Response(data=response_dict, status=status.HTTP_200_OK)
131 @action(detail=False, methods=["get"])
132 def filtered_statistics(self, request: Request, *args, **kwargs):
133 queryset = self.get_filtered_queryset(request)
135 result_dict = {}
136 result_dict["filtered_total_count"] = queryset.count()
138 return Response(data=result_dict)
141class SampleViewSetPlots(
142 SampleFilterMixin,
143 viewsets.GenericViewSet,
144 generics.mixins.ListModelMixin,
145 generics.mixins.RetrieveModelMixin,
146):
147 @staticmethod
148 def get_metadata_coverage(queryset):
149 """
150 Return a dict with counts of non-null values for each metadata field.
152 Performance optimized:
153 - Separate queries instead of massive aggregate()
154 - Leverages Django's query optimization and JOIN handling
155 - Caching of results for repeated queries
157 Example output:{
158 "metadata_coverage": {
159 "id": 17942,
160 "name": 17942,
161 "lineage": 17942,
162 "genome_completeness": 0,
163 "collection_date": 17942,
164 .....
165 "last_update_date": 17942,
166 "genomic_profiles": 17942,
167 "proteomic_profiles": 17941
168 }
169 }
170 """
171 # STEP 1: Cache check - return immediately if cached
172 queryset_sql = str(queryset.query)
173 cache_key = (
174 f"metadata_coverage:{hashlib.md5(queryset_sql.encode()).hexdigest()}"
175 )
177 cached_result = cache.get(cache_key)
178 if cached_result is not None:
179 return cached_result
181 # Initialize result dictionary
182 final_result = {}
184 # STEP 2: Count Sample table fields separately
185 # Instead of building 12+ annotations, query each field individually
186 # Django optimizes each query better than one massive aggregate()
188 for field in models.Sample._meta.get_fields():
189 if field.concrete and not field.is_relation:
190 field_name = field.name
192 try:
193 if field.get_internal_type() == "CharField":
194 # Count non-null AND non-empty strings
195 # Django will use the existing queryset filter + add this condition
196 count = (
197 queryset.filter(**{f"{field_name}__isnull": False})
198 .exclude(**{field_name: ""})
199 .values("id")
200 .distinct()
201 .count()
202 )
203 else:
204 # Count non-null values only
205 count = (
206 queryset.filter(**{f"{field_name}__isnull": False})
207 .values("id")
208 .distinct()
209 .count()
210 )
212 final_result[field_name] = count
213 except Exception as e:
214 LOGGER.error(f"Error counting field {field_name}: {e}")
215 final_result[field_name] = 0
217 # STEP 3: Count custom properties from Sample2Property table
218 # Query separately for each property using JOIN optimization
220 property_to_datatype = dict(
221 models.Property.objects.values_list("name", "datatype")
222 )
223 property_names = PropertyViewSet.get_custom_property_names()
225 for property_name in property_names:
226 datatype = property_to_datatype.get(property_name)
227 if not datatype:
228 LOGGER.warning(f"Property {property_name} not found")
229 continue
231 try:
232 # Django JOIN optimization: Sample -> Sample2Property -> Property
233 # Uses index on property__name and datatype column
234 count = (
235 queryset.filter(
236 properties__property__name=property_name,
237 **{f"properties__{datatype}__isnull": False},
238 )
239 .values("id")
240 .distinct()
241 .count()
242 )
244 final_result[property_name] = count
245 except Exception as e:
246 LOGGER.error(
247 f"Error counting property {property_name} (datatype: {datatype}): {e}"
248 )
249 final_result[property_name] = 0
251 # STEP 4: Count mutation profiles (genomic and proteomic)
252 # These queries are already optimized - keep as is
253 try:
254 # Count samples with nucleotide mutations
255 genomic_count = (
256 queryset.filter(
257 sequences__alignments__nucleotide_mutations__isnull=False
258 )
259 .values("id")
260 .distinct()
261 .count()
262 )
263 final_result["genomic_profiles"] = genomic_count
264 except Exception as e:
265 LOGGER.error(f"Error counting genomic profiles: {e}")
266 final_result["genomic_profiles"] = 0
268 try:
269 # Count samples with amino acid mutations
270 proteomic_count = (
271 queryset.filter(
272 sequences__alignments__amino_acid_mutations__isnull=False
273 )
274 .values("id")
275 .distinct()
276 .count()
277 )
278 final_result["proteomic_profiles"] = proteomic_count
279 except Exception as e:
280 LOGGER.error(f"Error counting proteomic profiles: {e}")
281 final_result["proteomic_profiles"] = 0
283 # STEP 5: Cache result for xx minutes and return
284 cache.set(cache_key, final_result, CACHE_OBJECT_TTL)
286 return final_result
288 def _get_samples_per_week(self, queryset):
289 """
290 Return a dict mapping calendar week to count, for each week between
291 the earliest and latest collection_date.
292 Weeks with zero records will be present with value 0.
293 """
294 weekly_qs = (
295 queryset.annotate(week=TruncWeek("collection_date"))
296 .values("week")
297 .annotate(count=Count("id", distinct=True))
298 .order_by("week")
299 )
301 result = {}
302 for item in weekly_qs:
303 year, week, _ = item["week"].isocalendar()
304 result[f"{year}-W{week:02}"] = item["count"]
306 return result
308 def _get_grouped_lineages_per_week(self, queryset):
309 """
310 Return a LIST of dicts, contianing counts and percentages per calendar week for
311 lineage groups (lineages truncated to two first segments),
312 covering every week between the earliest and latest record.
313 Weeks with zero records will be present with values 0.
314 """
315 # Generate cache key
316 queryset_sql = str(queryset.query)
317 cache_key = f"grouped_lineages_per_week:{hashlib.md5(queryset_sql.encode()).hexdigest()}"
319 # Try cache first
320 cached_result = cache.get(cache_key)
321 if cached_result is not None:
322 return cached_result
324 present_lineages = set(
325 queryset.exclude(lineage__isnull=True).values_list("lineage", flat=True)
326 )
327 lineage_to_group = {
328 lineage: ".".join(lineage.split(".")[:2]) for lineage in present_lineages
329 }
330 valid_groups = set(
331 models.Lineage.objects.filter(name__isnull=False).values_list(
332 "name", flat=True
333 )
334 )
335 lineage_cases = [
336 When(lineage=lineage, then=Value(group))
337 for lineage, group in lineage_to_group.items()
338 if group in valid_groups
339 ]
340 annotated_qs = queryset.annotate(
341 lineage_group=Case(
342 *lineage_cases,
343 default=Value("Unknown"),
344 output_field=CharField(),
345 ),
346 week=TruncWeek("collection_date"),
347 )
349 weekly_qs = annotated_qs.values("week", "lineage_group").annotate(
350 count=Count("id")
351 )
353 total_qs = annotated_qs.values("week").annotate(total_count=Count("id"))
355 # lookup for total counts
356 week_to_total = {entry["week"]: entry["total_count"] for entry in total_qs}
358 result = []
359 for item in weekly_qs:
360 year, week, _ = item["week"].isocalendar()
361 week_str = f"{year}-W{week:02}"
362 total = week_to_total.get(item["week"])
363 result.append(
364 {
365 "week": week_str,
366 "lineage_group": item["lineage_group"],
367 "count": item["count"],
368 "percentage": round(item["count"] / total * 100, 2),
369 }
370 )
372 result.sort(key=lambda x: x["week"])
374 # Cache for xx minutes
375 cache.set(cache_key, result, 60)
377 return result
379 def _get_custom_property_plot(self, queryset, sample_property):
380 # Determine if x_property and y_property are flexible or fixed
381 flexible_properties = PropertyViewSet.get_custom_property_names()
382 is_flexible = sample_property in flexible_properties
383 if sample_property == "":
384 result_dict = {}
385 # custom properties
386 elif is_flexible:
387 datatype = (
388 models.Property.objects.filter(name=sample_property)
389 .values_list("datatype", flat=True)
390 .first()
391 )
392 queryset = queryset.filter(properties__property__name=sample_property)
393 # the value_char holds the sequencing reason values
394 grouped_queryset = (
395 queryset.values(f"properties__{datatype}")
396 .annotate(total=Count("id", distinct=True))
397 .order_by("properties__value_varchar")
398 )
399 result_dict = {
400 item[f"properties__{datatype}"]: item["total"]
401 for item in grouped_queryset
402 }
403 # fixed sample table property
404 else:
405 grouped_queryset = (
406 queryset.values(sample_property)
407 .annotate(total=Count("id", distinct=True))
408 .order_by(sample_property)
409 )
410 result_dict = {
411 str(item[sample_property]): item["total"] for item in grouped_queryset
412 } # str required for properties in date format
414 return result_dict
416 @action(detail=False, methods=["get"])
417 def plot_samples_per_week(self, request: Request, *args, **kwargs):
418 queryset = self.get_filtered_queryset(request).filter(
419 collection_date__isnull=False
420 )
422 if not queryset.exists():
423 return Response(data=[])
424 else:
425 samples_per_week = self._get_samples_per_week(queryset)
426 return Response(data=list(samples_per_week.items()))
428 @action(detail=False, methods=["get"])
429 def plot_grouped_lineages_per_week(self, request: Request, *args, **kwargs):
430 queryset = self.get_filtered_queryset(request).filter(
431 collection_date__isnull=False
432 )
434 result_dict = {}
435 if not queryset.exists():
436 result_dict["grouped_lineages_per_week"] = {}
437 else:
438 result_dict["grouped_lineages_per_week"] = (
439 self._get_grouped_lineages_per_week(queryset)
440 )
442 return Response(data=result_dict)
444 @action(detail=False, methods=["get"])
445 def plot_metadata_coverage(self, request: Request, *args, **kwargs):
446 queryset = self.get_filtered_queryset(request)
447 result_dict = {}
448 result_dict["metadata_coverage"] = self.get_metadata_coverage(queryset)
449 return Response(data=result_dict)
451 @action(detail=False, methods=["get"])
452 def plot_custom(self, request: Request, *args, **kwargs):
453 queryset = self.get_filtered_queryset(request)
454 sample_property = request.query_params["property"]
456 result_dict = {}
457 result_dict[sample_property] = self._get_custom_property_plot(
458 queryset, sample_property
459 )
460 return Response(data=result_dict)
462 @action(detail=False, methods=["get"])
463 def plot_custom_xy(self, request: Request, *args, **kwargs):
464 """
465 API call to plot data based on two properties: x and y categories.
466 Handles both flexible properties (sample2property table) and fixed sample table properties.
467 Returns:
468 - For string-type y property: dict with x categories as keys and lists of y categories with counts.
469 - For number-type y property: dict with x categories as keys and lists of y numbers.
470 """
471 queryset = self.get_filtered_queryset(request)
472 x_property = request.query_params.get("x_property")
473 y_property = request.query_params.get("y_property")
475 if not x_property or not y_property:
476 return Response(
477 {"detail": "Both x_property and y_property must be provided."},
478 status=status.HTTP_400_BAD_REQUEST,
479 )
481 # Determine if x_property and y_property are flexible or fixed
482 flexible_properties = PropertyViewSet.get_custom_property_names()
483 x_is_flexible = x_property in flexible_properties
484 y_is_flexible = y_property in flexible_properties
486 result_dict = {}
488 if y_is_flexible:
489 # Handle flexible y_property
490 y_datatype = (
491 models.Property.objects.filter(name=y_property)
492 .values_list("datatype", flat=True)
493 .first()
494 )
496 if not y_datatype:
497 return Response(
498 {"detail": f"Property {y_property} not found."},
499 status=status.HTTP_400_BAD_REQUEST,
500 )
502 if x_is_flexible:
503 # Both x_property and y_property are flexible
504 grouped_queryset = (
505 queryset.filter(
506 properties__property__name__in=[x_property, y_property],
507 properties__value_varchar__isnull=False,
508 **{f"properties__{y_datatype}__isnull": False},
509 )
510 .values(
511 "properties__value_varchar",
512 f"properties__{y_datatype}",
513 )
514 .annotate(total=Count("id", distinct=True))
515 .order_by("properties__value_varchar", f"properties__{y_datatype}")
516 )
517 for item in grouped_queryset:
518 x_cat = str(item["properties__value_varchar"])
519 y_cat = str(item[f"properties__{y_datatype}"])
520 count = item["total"]
521 if x_cat not in result_dict:
522 result_dict[x_cat] = []
523 result_dict[x_cat].append({y_cat: count})
525 else:
526 # x_property is fixed, y_property is flexible
527 grouped_queryset = (
528 queryset.filter(
529 **{f"{x_property}__isnull": False},
530 properties__property__name=y_property,
531 **{f"properties__{y_datatype}__isnull": False},
532 )
533 .values(x_property, f"properties__{y_datatype}")
534 .annotate(total=Count("id", distinct=True))
535 .order_by(x_property, f"properties__{y_datatype}")
536 )
537 for item in grouped_queryset:
538 x_cat = str(item[x_property])
539 y_cat = str(item[f"properties__{y_datatype}"])
540 count = item["total"]
541 if x_cat not in result_dict:
542 result_dict[x_cat] = []
543 result_dict[x_cat].append({y_cat: count})
545 else:
546 # Handle fixed y_property
547 if x_is_flexible:
548 # x_property is flexible, y_property is fixed
549 grouped_queryset = (
550 queryset.filter(
551 properties__property__name=x_property,
552 **{f"{y_property}__isnull": False},
553 properties__value_varchar__isnull=False,
554 )
555 .values("properties__value_varchar", y_property)
556 .annotate(total=Count("id", distinct=True))
557 .order_by("properties__value_varchar", y_property)
558 )
559 for item in grouped_queryset:
560 x_cat = str(item["properties__value_varchar"])
561 y_cat = str(item[y_property])
562 count = item["total"]
563 if x_cat not in result_dict:
564 result_dict[x_cat] = []
565 result_dict[x_cat].append({y_cat: count})
567 else:
568 # Both x_property and y_property are fixed
569 grouped_queryset = (
570 queryset.filter(
571 **{f"{x_property}__isnull": False},
572 **{f"{y_property}__isnull": False},
573 )
574 .values(x_property, y_property)
575 .annotate(total=Count("id", distinct=True))
576 .order_by(x_property, y_property)
577 )
578 for item in grouped_queryset:
579 x_cat = str(item[x_property])
580 y_cat = str(item[y_property])
581 count = item["total"]
582 if x_cat not in result_dict:
583 result_dict[x_cat] = []
584 result_dict[x_cat].append({y_cat: count})
586 return Response(data=result_dict)