Login

aggregate filter

Author:
atombrella
Posted:
January 25, 2017
Language:
Python
Version:
1.8
Score:
3 (after 3 ratings)

Makes it possible to add a filtering condition directly after the aggregate function (or possible, aggregate(expression) WITHIN GROUP (ordering clause). This is mostly useful if the annotation has two or more expressions, so it's possible to compare the result with and without the applied filter; it's more compact than using Case. It's suggested to add values to the queryset to get a proper group by.

Usage example:

books = Book.objects.values('publisher__name').annotate( count=Count('*'), filtercount=Filter(expression=Count('publisher__name'), condition=Q(rating__gte=5)) )

Supported on Postgresql 9.4+. Possible other third-party backends.

 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
from django.db.models import Expression

class Filter(Expression):
    template = '%(expression)s FILTER (WHERE %(condition)s)'

    def __init__(self, expression, condition, output_field=None):
        if not expression.contains_aggregate:
            raise TypeError('Expression must either be an aggregate function or contain an aggregate function')

        if not hasattr(condition, 'resolve_expression'):
            raise TypeError('Condition must be a class defining resolve_expression')

        super().__init__(output_field=output_field)
        self.source_expression = self._parse_expressions(expression)[0]
        self.condition = condition
        if not getattr(self.source_expression, 'contains_aggregate', False):
            raise FieldError('Window function expressions must be aggregate functions')

    def resolve_expression(self, query=None, allow_joins=True, reuse=None, summarize=False, for_save=False):
        c = self.copy()
        c.source_expression = self.source_expression.resolve_expression(query, allow_joins, reuse, summarize, for_save)
        c.condition = self.condition.resolve_expression(query, allow_joins, reuse, summarize, for_save)
        return c

    def _resolve_output_field(self):
        if self._output_field is None:
            self._output_field = self.source_expression.output_field

    def copy(self):
        clone = super().copy()
        clone.source_expression = self.source_expression.copy()
        clone.condition = copy.copy(self.condition)
        return clone

    def as_sql(self, compiler, connection):
        connection.ops.check_expression_support(self)
        params = []
        condition_sql, condition_params = compiler.compile(self.condition)
        params.extend(condition_params)
        expr_sql, expr_params = compiler.compile(self.source_expression)
        condition_params.extend(expr_params)

        return self.template % {
            'expression': expr_sql,
            'condition': condition_sql,
        }, params

    def get_source_expressions(self):
        return self.source_expression, self.condition

    def set_source_expressions(self, exprs):
        self.source_expression, self.condition = exprs[0], exprs[1]

    def get_group_by_cols(self):
        return []

    def __str__(self):
        return self.template % {
            'expression': str(self.source_expression),
            'condition': str(self.condition),
        }

    def __repr__(self):
        return '<%s: %s>' % (self.__class__.__name__, self)

More like this

  1. Month / Year SelectDateWidget based on django SelectDateWidget by pierreben 3 months ago
  2. Python Django CRUD Example Tutorial by tuts_station 3 months, 2 weeks ago
  3. Browser-native date input field by kytta 5 months ago
  4. Generate and render HTML Table by LLyaudet 5 months, 1 week ago
  5. My firs Snippets by GutemaG 5 months, 1 week ago

Comments

Please login first before commenting.