Dry ORM

Code

Result

no cache

from django.db import models
from django.db.models import Count, Q, Subquery, OuterRef, F, Value, FloatField, Case, When, ExpressionWrapper
import uuid

class OuterObj(models.Model):
    low_count = models.IntegerField(default=0)
    low_count_filtered = models.IntegerField(default=0)
    percentage = models.FloatField(default=0.0)

class MidObj(models.Model):
    outer = models.ForeignKey(OuterObj, on_delete=models.CASCADE, related_name='mids')

class LowObj(models.Model):
    midobj = models.ForeignKey(MidObj, on_delete=models.CASCADE, related_name='lows')
    uuid = models.UUIDField(default=uuid.uuid4, unique=True)
    prop = models.CharField(max_length=50)

def update_outer(low_uuids):
    # Build a correlated subquery that computes totals over ALL LowObjs for each OuterObj
    base = (
        OuterObj.objects
        .filter(pk=OuterRef('pk'))
        .annotate(
            tcount=Count('mids__lows__id'),
            fcount=Count('mids__lows__id', filter=Q(mids__lows__prop="value")),
        )
    )

percent_qs = base.annotate(
        percent=Case(
            When(tcount=0, then=Value(0.0)),
            default=ExpressionWrapper(100.0 * F('fcount') / F('tcount'), output_field=FloatField()),
        )
    )

# Update only Outers touched by the provided LowObj UUIDs
    updated = (
        OuterObj.objects
        .filter(mids__lows__uuid__in=low_uuids)
        .update(
            low_count=Subquery(base.values('tcount')[:1]),
            low_count_filtered=Subquery(base.values('fcount')[:1]),
            percentage=Subquery(percent_qs.values('percent')[:1]),
        )
    )
    return updated

def run():
    # Bootstrap sample data
    o1 = OuterObj.objects.create()
    o2 = OuterObj.objects.create()
    o3 = OuterObj.objects.create()  # will have no lows

m11 = MidObj.objects.create(outer=o1)
    m12 = MidObj.objects.create(outer=o1)
    m21 = MidObj.objects.create(outer=o2)

lows = []
    lows.append(LowObj.objects.create(midobj=m11, prop="value"))
    lows.append(LowObj.objects.create(midobj=m11, prop="other"))
    lows.append(LowObj.objects.create(midobj=m12, prop="value"))
    lows.append(LowObj.objects.create(midobj=m12, prop="other"))
    lows.append(LowObj.objects.create(midobj=m21, prop="other"))

# Choose a few LowObj UUIDs that should trigger updates for o1 and o2
    trigger_uuids = [lows[0].uuid, lows[4].uuid]

print('Before update:')
    for o in OuterObj.objects.order_by('id'):
        print(f'Outer {o.id}: low_count={o.low_count}, low_count_filtered={o.low_count_filtered}, percentage={o.percentage}')

updated_rows = update_outer(trigger_uuids)
    print(f'Updated rows: {updated_rows}')

print('After update:')
    for o in OuterObj.objects.order_by('id'):
        print(f'Outer {o.id}: low_count={o.low_count}, low_count_filtered={o.low_count_filtered}, percentage={o.percentage}')

Output

No output.

Queries

Queries Q (0)

No queries.