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 
32
33
34
35
36
def value_queryset(self, sub_domain: SubDomain | None = None) -> QuerySet[StatisticValue]:
    queryset = self.obj.values.select_related('sub_domain')
    if sub_domain is not None:
        queryset = queryset.for_sub_domain(sub_domain)
    return queryset

values_for_date

Source code in django_spire/metric/domain/statistic/services/transformation_service.py 
38
39
40
41
42
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 
44
45
46
47
def total_for_date(
    self, value_date: date_type | None = None, *, sub_domain: SubDomain | None = None
) -> Decimal:
    return self.values_for_date(value_date, sub_domain=sub_domain).total()

daily_summary

Source code in django_spire/metric/domain/statistic/services/transformation_service.py 
49
50
51
52
53
54
55
56
57
58
59
60
def daily_summary(
    self, start_date: date_type, end_date: date_type, *, sub_domain: SubDomain | None = None
) -> dict[date_type, Decimal]:
    summary: dict[date_type, Decimal] = {}

    for value in (
        self.value_queryset(sub_domain).date_range(start_date, end_date).order_by('timestamp')
    ):
        day = timezone.localtime(value.timestamp).date()
        summary[day] = summary.get(day, Decimal(0)) + value.value

    return dict(sorted(summary.items()))

total_between

Source code in django_spire/metric/domain/statistic/services/transformation_service.py 
62
63
64
65
def total_between(
    self, start_date: date_type, end_date: date_type, *, sub_domain: SubDomain | None = None
) -> Decimal:
    return self.value_queryset(sub_domain).date_range(start_date, end_date).total()

interval_bounds

Source code in django_spire/metric/domain/statistic/services/transformation_service.py 
67
68
69
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 
71
72
73
74
75
76
77
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 
79
80
81
82
def total_for_interval(
    self, value_date: date_type | None = None, *, sub_domain: SubDomain | None = None
) -> Decimal:
    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 
84
85
86
87
88
89
90
91
92
93
94
def interval_summary(
    self, start_date: date_type, end_date: date_type, *, sub_domain: SubDomain | None = None
) -> dict[date_type, Decimal]:
    summary: dict[date_type, Decimal] = {}

    for value in self.value_queryset(sub_domain).date_range(start_date, end_date):
        day = timezone.localtime(value.timestamp).date()
        bucket_start, _ = interval_range(self.obj.interval, day)
        summary[bucket_start] = summary.get(bucket_start, Decimal(0)) + value.value

    return dict(sorted(summary.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