Skip to content

transformation_service

django_spire.metric.domain.statistic.services.transformation_service

StatisticGroupTransformationService

Bases: BaseDjangoModelService['StatisticGroup']

Source code in django_spire/contrib/constructor/constructor.py 
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
def __init__(self, obj: Any = None):
    self._obj_type_name: str = str(next(iter(self.__class__.__annotations__.values()))).split(
        '.'
    )[-1]

    if obj is None:
        return

    self._obj_mro_type_names = [cls.__name__ for cls in obj.__class__.__mro__]

    if self._obj_type_name not in self._obj_mro_type_names:
        message = f'{self.__class__.__name__} was instantiated with obj type "{obj.__class__.__name__}" and failed as it was expecting "{self._obj_type_name}".'
        raise ConstructorError(message)

    self._obj_type: type[TypeAny] = obj.__class__

    if self._obj_type is None or self._obj_type is ...:
        message = f'{self.__class__.__name__} top class attribute must have an annotated type.'
        raise ConstructorError(message)

    self.obj: TypeAny = obj

    if ABC not in self.__class__.__bases__:
        if not self._obj_is_valid:
            message = f'{self._obj_type_name} failed to validate on {self.__class__.__name__}'
            raise ConstructorError(message)

    self.__post_init__()

obj instance-attribute

StatisticTransformationService

Bases: BaseDjangoModelService['Statistic']

Source code in django_spire/contrib/constructor/constructor.py 
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
def __init__(self, obj: Any = None):
    self._obj_type_name: str = str(next(iter(self.__class__.__annotations__.values()))).split(
        '.'
    )[-1]

    if obj is None:
        return

    self._obj_mro_type_names = [cls.__name__ for cls in obj.__class__.__mro__]

    if self._obj_type_name not in self._obj_mro_type_names:
        message = f'{self.__class__.__name__} was instantiated with obj type "{obj.__class__.__name__}" and failed as it was expecting "{self._obj_type_name}".'
        raise ConstructorError(message)

    self._obj_type: type[TypeAny] = obj.__class__

    if self._obj_type is None or self._obj_type is ...:
        message = f'{self.__class__.__name__} top class attribute must have an annotated type.'
        raise ConstructorError(message)

    self.obj: TypeAny = obj

    if ABC not in self.__class__.__bases__:
        if not self._obj_is_valid:
            message = f'{self._obj_type_name} failed to validate on {self.__class__.__name__}'
            raise ConstructorError(message)

    self.__post_init__()

obj instance-attribute

value_queryset

Source code in django_spire/metric/domain/statistic/services/transformation_service.py 
38
39
40
41
42
43
44
45
def value_queryset(self, sub_domain: SubDomain | None = None) -> QuerySet[StatisticValue]:
    if self.obj.is_deleted:
        return self.obj.values.none()

    queryset = self.obj.values.select_related('sub_domain')
    if sub_domain is not None:
        queryset = queryset.for_sub_domain(sub_domain)
    return queryset

aggregate

Source code in django_spire/metric/domain/statistic/services/transformation_service.py 
47
48
49
50
51
def aggregate(self, queryset: QuerySet[StatisticValue]) -> Decimal:
    if self.obj.value_type == StatisticValueTypeChoices.PERCENTAGE:
        return queryset.average()

    return queryset.total()

values_for_date

Source code in django_spire/metric/domain/statistic/services/transformation_service.py 
53
54
55
56
57
def values_for_date(
    self, value_date: date_type | None = None, *, sub_domain: SubDomain | None = None
) -> QuerySet[StatisticValue]:
    value_date = value_date or timezone.localdate()
    return self.value_queryset(sub_domain).for_date(value_date).order_by('timestamp')

total_for_date

Source code in django_spire/metric/domain/statistic/services/transformation_service.py 
59
60
61
62
def total_for_date(
    self, value_date: date_type | None = None, *, sub_domain: SubDomain | None = None
) -> Decimal:
    return self.aggregate(self.values_for_date(value_date, sub_domain=sub_domain))

daily_summary

Source code in django_spire/metric/domain/statistic/services/transformation_service.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
def daily_summary(
    self, start_date: date_type, end_date: date_type, *, sub_domain: SubDomain | None = None
) -> dict[date_type, Decimal]:
    aggregate = Avg if self.obj.value_type == StatisticValueTypeChoices.PERCENTAGE else Sum

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

    return {row['day']: row['total'] for row in rows}

total_between

Source code in django_spire/metric/domain/statistic/services/transformation_service.py 
80
81
82
83
84
def total_between(
    self, start_date: date_type, end_date: date_type, *, sub_domain: SubDomain | None = None
) -> Decimal:
    queryset = self.value_queryset(sub_domain).date_range(start_date, end_date)
    return self.aggregate(queryset)

interval_bounds

Source code in django_spire/metric/domain/statistic/services/transformation_service.py 
86
87
88
def interval_bounds(self, value_date: date_type | None = None) -> tuple[date_type, date_type]:
    value_date = value_date or timezone.localdate()
    return interval_range(self.obj.interval, value_date)

values_for_interval

Source code in django_spire/metric/domain/statistic/services/transformation_service.py 
90
91
92
93
94
95
96
def values_for_interval(
    self, value_date: date_type | None = None, *, sub_domain: SubDomain | None = None
) -> QuerySet[StatisticValue]:
    start_date, end_date = self.interval_bounds(value_date)
    return (
        self.value_queryset(sub_domain).date_range(start_date, end_date).order_by('timestamp')
    )

total_for_interval

Source code in django_spire/metric/domain/statistic/services/transformation_service.py 
 98
 99
100
101
102
103
104
105
106
def total_for_interval(
    self, value_date: date_type | None = None, *, sub_domain: SubDomain | None = None
) -> Decimal:
    if self.obj.value_type == StatisticValueTypeChoices.PERCENTAGE:
        value_date = value_date or timezone.localdate()
        window_days = percentage_moving_window_days(self.obj.interval)
        return self.value_queryset(sub_domain).moving_window_average(value_date, window_days)

    return self.values_for_interval(value_date, sub_domain=sub_domain).total()

interval_summary

Source code in django_spire/metric/domain/statistic/services/transformation_service.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
def interval_summary(
    self, start_date: date_type, end_date: date_type, *, sub_domain: SubDomain | None = None
) -> dict[date_type, Decimal]:
    is_percentage = self.obj.value_type == StatisticValueTypeChoices.PERCENTAGE

    queryset = (
        self.value_queryset(sub_domain)
        .date_range(start_date, end_date)
        .annotate(day=TruncDate('timestamp', tzinfo=timezone.get_current_timezone()))
        .values('day')
        .annotate(total=Sum('value'))
    )
    if is_percentage:
        queryset = queryset.annotate(count=Count('pk'))

    totals: dict[date_type, Decimal] = {}
    counts: dict[date_type, int] = {}
    for row in queryset:
        bucket_start, _ = interval_range(self.obj.interval, row['day'])
        totals[bucket_start] = totals.get(bucket_start, Decimal(0)) + row['total']
        if is_percentage:
            counts[bucket_start] = counts.get(bucket_start, 0) + row['count']

    if is_percentage:
        return dict(sorted((start, totals[start] / counts[start]) for start in totals))

    return dict(sorted(totals.items()))

StatisticValueTransformationService

Bases: BaseDjangoModelService['StatisticValue']

Source code in django_spire/contrib/constructor/constructor.py 
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
def __init__(self, obj: Any = None):
    self._obj_type_name: str = str(next(iter(self.__class__.__annotations__.values()))).split(
        '.'
    )[-1]

    if obj is None:
        return

    self._obj_mro_type_names = [cls.__name__ for cls in obj.__class__.__mro__]

    if self._obj_type_name not in self._obj_mro_type_names:
        message = f'{self.__class__.__name__} was instantiated with obj type "{obj.__class__.__name__}" and failed as it was expecting "{self._obj_type_name}".'
        raise ConstructorError(message)

    self._obj_type: type[TypeAny] = obj.__class__

    if self._obj_type is None or self._obj_type is ...:
        message = f'{self.__class__.__name__} top class attribute must have an annotated type.'
        raise ConstructorError(message)

    self.obj: TypeAny = obj

    if ABC not in self.__class__.__bases__:
        if not self._obj_is_valid:
            message = f'{self._obj_type_name} failed to validate on {self.__class__.__name__}'
            raise ConstructorError(message)

    self.__post_init__()

obj instance-attribute