Skip to content

querysets

django_spire.metric.domain.statistic.querysets

StatisticGroupQuerySet

Bases: HistoryQuerySet, SearchQuerySetMixin

bulk_filter

Source code in django_spire/metric/domain/statistic/querysets.py 
60
61
62
63
64
65
66
67
def bulk_filter(self, filter_data: dict) -> QuerySet[StatisticGroup]:
    queryset = self

    search = filter_data.get('search', '')
    if search:
        queryset = queryset.search(search)

    return queryset

StatisticQuerySet

Bases: HistoryQuerySet, SearchQuerySetMixin

for_key

Source code in django_spire/metric/domain/statistic/querysets.py 
71
72
def for_key(self, key: str) -> QuerySet[Statistic]:
    return self.filter(key=key)

bulk_filter

Source code in django_spire/metric/domain/statistic/querysets.py 
74
75
76
77
78
79
80
81
def bulk_filter(self, filter_data: dict) -> QuerySet[Statistic]:
    queryset = self

    search = filter_data.get('search', '')
    if search:
        queryset = queryset.search(search)

    return queryset

StatisticValueQuerySet

Bases: QuerySet

for_date

Source code in django_spire/metric/domain/statistic/querysets.py 
85
86
87
88
89
def for_date(self, value_date: date) -> QuerySet[StatisticValue]:
    return self.filter(
        timestamp__gte=local_day_start(value_date),
        timestamp__lt=local_day_start(value_date + timedelta(days=1)),
    )

date_range

Source code in django_spire/metric/domain/statistic/querysets.py 
91
92
93
94
95
def date_range(self, start_date: date, end_date: date) -> QuerySet[StatisticValue]:
    return self.filter(
        timestamp__gte=local_day_start(start_date),
        timestamp__lt=local_day_start(end_date + timedelta(days=1)),
    )

for_reference

Source code in django_spire/metric/domain/statistic/querysets.py 
97
98
def for_reference(self, reference: str) -> QuerySet[StatisticValue]:
    return self.filter(reference=reference)

for_reference_pattern

Source code in django_spire/metric/domain/statistic/querysets.py 
100
101
102
103
104
def for_reference_pattern(self, pattern: str) -> QuerySet[StatisticValue]:
    if pattern == '':
        return self

    return self.filter(reference_pattern_q(pattern))

for_reference_patterns

Source code in django_spire/metric/domain/statistic/querysets.py 
106
107
108
109
110
111
112
113
114
115
def for_reference_patterns(self, patterns: list[str]) -> QuerySet[StatisticValue]:
    if not patterns:
        return self

    filters = [reference_pattern_q(pattern) for pattern in patterns if pattern != '']

    if not filters:
        return self

    return self.filter(reduce(or_, filters))

for_sub_domain

Source code in django_spire/metric/domain/statistic/querysets.py 
117
118
def for_sub_domain(self, sub_domain: SubDomain) -> QuerySet[StatisticValue]:
    return self.filter(sub_domain=sub_domain)

total

Source code in django_spire/metric/domain/statistic/querysets.py 
120
121
def total(self) -> Decimal:
    return self.aggregate(total=Sum('value'))['total'] or Decimal(0)

average

Source code in django_spire/metric/domain/statistic/querysets.py 
123
124
def average(self) -> Decimal:
    return self.aggregate(total=Avg('value'))['total'] or Decimal(0)

series_points

Source code in django_spire/metric/domain/statistic/querysets.py 
126
127
128
129
130
131
132
133
134
135
def series_points(self, start_date: date, end_date: date) -> list[tuple[date, Decimal]]:
    rows = (
        self.date_range(start_date, end_date)
        .annotate(day=TruncDate('timestamp', tzinfo=timezone.get_current_timezone()))
        .values('day')
        .annotate(total=Sum('value'))
        .order_by('day')
    )

    return [(row['day'], Decimal(row['total'])) for row in rows]

breakdown

Source code in django_spire/metric/domain/statistic/querysets.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
def breakdown(
    self, start_date: date, end_date: date, *, average: bool = False
) -> list[tuple[str, Decimal]]:
    aggregate = Avg('value') if average else Sum('value')

    rows = (
        self.date_range(start_date, end_date)
        .values('reference')
        .annotate(total=aggregate)
        .order_by('reference')
    )

    return [(row['reference'] or 'Unassigned', Decimal(row['total'])) for row in rows]

daily_averages

Source code in django_spire/metric/domain/statistic/querysets.py 
151
152
153
154
155
156
157
158
159
160
def daily_averages(self, start_date: date, end_date: date) -> list[tuple[date, Decimal]]:
    rows = (
        self.date_range(start_date, end_date)
        .annotate(day=TruncDate('timestamp', tzinfo=timezone.get_current_timezone()))
        .values('day')
        .annotate(total=Avg('value'))
        .order_by('day')
    )

    return [(row['day'], Decimal(row['total'])) for row in rows]

moving_window_average

Source code in django_spire/metric/domain/statistic/querysets.py 
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
def moving_window_average(self, end_date: date, window_days: int) -> Decimal:
    start_date = end_date - timedelta(days=window_days - 1)

    rows = (
        self.date_range(start_date, end_date)
        .annotate(day=TruncDate('timestamp', tzinfo=timezone.get_current_timezone()))
        .values('day')
        .annotate(total=Avg('value'))
    )

    values = [row['total'] for row in rows]
    if not values:
        return Decimal(0)

    return sum(values, Decimal(0)) / len(values)

contains_wildcard

Source code in django_spire/metric/domain/statistic/querysets.py 
28
29
def contains_wildcard(pattern: str) -> bool:
    return '%' in pattern or '_' in pattern

pattern_to_regex

Source code in django_spire/metric/domain/statistic/querysets.py 
32
33
def pattern_to_regex(pattern: str) -> str:
    return re.escape(pattern).replace(r'%', '.*').replace(r'_', '.')

reference_matches

Source code in django_spire/metric/domain/statistic/querysets.py 
36
37
38
39
40
def reference_matches(pattern: str, reference: str) -> bool:
    if not contains_wildcard(pattern):
        return pattern == reference

    return re.fullmatch(pattern_to_regex(pattern), reference) is not None

reference_pattern_q

Source code in django_spire/metric/domain/statistic/querysets.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
def reference_pattern_q(pattern: str) -> Q:
    if not contains_wildcard(pattern):
        return Q(reference=pattern)

    if pattern.endswith('%') and not contains_wildcard(pattern[:-1]):
        return Q(reference__startswith=pattern[:-1])

    if pattern.startswith('%') and not contains_wildcard(pattern[1:]):
        return Q(reference__endswith=pattern[1:])

    if pattern.startswith('%') and pattern.endswith('%') and not contains_wildcard(pattern[1:-1]):
        return Q(reference__contains=pattern[1:-1])

    return Q(reference__regex=f'^{pattern_to_regex(pattern)}$')