python / expert
Snippet
Django Aggregation with Conditional Expressions and Annotations
Django's aggregation framework combined with conditional expressions enables complex analytical queries directly in the ORM. This example demonstrates calculating revenue statistics using subqueries and OuterRef for correlated data, conditional aggregation with filter parameters to Count/Sum, and Case/When expressions for performance scoring. The techniques enable building sophisticated reporting features without raw SQL.
snippet.py
python
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
from django.db.models import (Sum, Avg, Count, Max, Min, F, Q, Case, When,IntegerField, DecimalField, Value, OuterRef, Subquery)from django.db.models.functions import Coalesce, Round, TruncMonthfrom django.db import transactionfrom decimal import Decimalclass SalesQuerySet(models.QuerySet):def with_revenue_stats(self, start_date=None, end_date=None):"""Calculate revenue statistics with conditional month grouping."""queryset = self.filter(is_completed=True)if start_date:queryset = queryset.filter(created_at__gte=start_date)if end_date:queryset = queryset.filter(created_at__lte=end_date)return queryset.annotate(total_revenue=Coalesce(Sum(F('items__quantity') * F('items__unit_price')),Value(0)),average_order_value=Round(Avg(F('items__quantity') * F('items__unit_price')),2),total_items_sold=Sum('items__quantity'),unique_customers=Count('customer', distinct=True),month=TruncMonth('created_at')).values('month', 'total_revenue', 'average_order_value','total_items_sold', 'unique_customers').order_by('month')def top_products_by_revenue(self, limit=10):"""Find top selling products with revenue calculation."""from .models import OrderItemrevenue_subquery = OrderItem.objects.filter(product=OuterRef('pk')).annotate(item_revenue=F('quantity') * F('unit_price')).values('product').annotate(total=Sum('item_revenue')).values('total')return self.annotate(total_revenue=Subquery(revenue_subquery, output_field=DecimalField()),units_sold=Subquery(OrderItem.objects.filter(product=OuterRef('pk')).values('product').annotate(total=Sum('quantity')).values('total'),output_field=IntegerField())).filter(is_active=True, units_sold__gt=0).order_by('-total_revenue')[:limit]def sales_performance_by_category(self):"""Aggregate sales metrics grouped by category with conditional logic."""return self.values('category__name').annotate(category_name=F('category__name'),total_products=Count('id'),active_products=Count('id', filter=Q(is_active=True)),out_of_stock=Count('id', filter=Q(stock_quantity=0)),low_stock=Count('id', filter=Q(stock_quantity__gt=0, stock_quantity__lte=10)),total_inventory_value=Sum(F('stock_quantity') * F('price')),average_price=Round(Avg('price'), 2),price_range=Max('price') - Min('price'),performance_score=Case(When(units_sold__gte=100, then=Value('excellent')),When(units_sold__gte=50, then=Value('good')),When(units_sold__gte=10, then=Value('average')),default=Value('poor'),output_field=IntegerField())).order_by('-total_inventory_value')
django
Breakdown
1
from django.db.models import Sum, Avg, Count, F, Case, When...
Import aggregation and conditional expression classes
2
Coalesce(Sum(...), Value(0))
Return 0 instead of None when no records match
3
Subquery(revenue_subquery, output_field=DecimalField())
Correlated subquery to calculate per-product revenue
4
OuterRef('pk')
Reference outer query field from within subquery
5
Count('id', filter=Q(is_active=True))
Conditional count using Q object filter
6
performance_score=Case(When(units_sold__gte=100, then=Value('excellent'))...)
Case/When for conditional annotation with string output
7
TruncMonth('created_at')
Group by calendar month for time-series analysis