Django: Unpack argument list for use in aggregate query
I am attempting to create a semi-dynamic aggregate function that will return the sums of all fields within a list. The assumption is that running get_query_set() will return a filtered query that contains all the fields in the list and some others that may not play so well with a Sum aggregate (date fields, char fields, Foreign Keys, etc...)
Best examples I've come up with so far are below, this is largely a python question with Django specific usage though my Python-Fu is not the strongest yet...
Works
qs = cl.get_query_set().aggregate(Sum('permits_submitted'), Sum('permits_pulled'), Sum('permits_posted'))
return: {'permits_pulled__sum': 5772, 'permits_posted__sum': 6723, 'permits_submitted__sum': 7276}
Does not work
qs = cl.get_query_set().aggregate(Sum('permits_submitted')).aggregate(Sum('permits_pulled'))
return: error
qs = cl.get_query_set().aggregate(Sum('permits_submitted', 'permits_pulled', Sum('permits_posted'))
return: error
Does not work - presents the idea
tuple = (
'permits_submitted',
'permits_pulled',
'permits_posted',
)
qs = cl.get_query_set()
for field in tuple:
qs.aggregate(Sum(field))
r开发者_StackOverflow中文版eturn: error
qs = cl.get_query_set()
qs.aggregate(*[Sum(field) for field in tuple])
return:
[<permit_runner: User's report for 2010-02-18>, <permit_runner: User's report for 2010-02-19>, '...(remaining elements truncated)...']
(this is the same as the return without aggregation)
WORKS
qs = cl.get_query_set()
qs = qs.aggregate(*[Sum(field) for field in tuple])
had missed defining qs = when adding the aggregation - helps to take a break for a few minutes and look fresh
Since this work, I put that as an answer, so people will find it easily when googling:
qs = cl.get_query_set()
qs = qs.aggregate(*[Sum(field) for field in tuple])
Creating dynamic aggregations (or dynamic annotations)
from typing import Dict
from django.db.models import Q, Sum
# 1. Prepare your querysets
years = year_allowances.values_list("year", flat=True)
time_aways = TimeAway.objects.filter(
sequence__pk__in=sequences.values_list("pk", flat=True)
).actual_paid_sick_leave()
# 2. Define your individual aggregate expression.
def get_aggregate(key) -> Dict[str, Sum]:
return {
str(key): Sum(F('actual_paid_sick_leave'), filter=Q(local_timezone_start__year=key))
}
# 3. Create the dictionary of aggregate expressions.
aggregate_expressions = {}
ds = [{**get_aggregate(year)} for year in years]
for d in ds:
aggregate_expressions.update(d)
# 4. Create your aggregations.
x = time_aways.aggregate(**aggregate_expressions)
>> x = {'2021': datetime.timedelta(0), '2022': datetime.timedelta(days=5)}
精彩评论