Skip to content

tracking_service

django_spire.metric.domain.statistic.services.tracking_service

logger = logging.getLogger(__name__) module-attribute

WRITE_TIMEOUT_MS = 2000 module-attribute

DELETE_BATCH_SIZE = 5000 module-attribute

StatisticTrackingService

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

track

Source code in django_spire/metric/domain/statistic/services/tracking_service.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
def track(
    self, sub_domain: SubDomain, *, reference: str = 'page_click'
) -> StatisticValue | None:
    if sub_domain.domain_id != self.obj.group.domain_id:
        logger.warning(
            'Sub-domain %s does not belong to domain %s', sub_domain, self.obj.group.domain
        )
        return None

    try:
        self._apply_write_timeout()

        with transaction.atomic():
            return self.obj.services.processor.increment(
                reference=reference, sub_domain=sub_domain
            )
    except Exception:
        logger.warning('Statistic tracking failed', exc_info=True)
        return None
    finally:
        self._reset_write_timeout()

trim

Source code in django_spire/metric/domain/statistic/services/tracking_service.py 
52
53
54
55
56
57
58
def trim(self, sub_domain: SubDomain, reference: str, *, max_values: int | None = None) -> int:
    cap = max_values or getattr(settings, 'DJANGO_SPIRE_METRIC_TRACKING_VALUES_MAX', 1000)

    values = self.obj.values.filter(sub_domain=sub_domain, reference=reference)
    retained_pks = list(values.order_by('-timestamp').values_list('pk', flat=True)[:cap])

    return self._delete_batch(values, retained_pks)

track_many classmethod

Source code in django_spire/metric/domain/statistic/services/tracking_service.py 
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
@classmethod
def track_many(cls, references: list[str]) -> None:
    from django_spire.metric.domain.models import SubDomain  # noqa: PLC0415
    from django_spire.metric.domain.statistic.models import (  # noqa: PLC0415
        Statistic,
        StatisticValue,
    )

    statistic_key = settings.DJANGO_SPIRE_INTERNAL_METRIC_STATISTIC_KEY
    sub_domain_key = settings.DJANGO_SPIRE_INTERNAL_METRIC_SUB_DOMAIN_KEY

    if (not statistic_key) or (not sub_domain_key):
        return

    statistic = (
        Statistic.objects.for_key(statistic_key)
        .active()
        .not_deleted()
        .select_related('group')
        .first()
    )
    sub_domain = SubDomain.objects.for_key(sub_domain_key).active().not_deleted().first()

    if statistic is None or sub_domain is None:
        logger.debug('Statistic tracking target not found')
        return

    if sub_domain.domain_id != statistic.group.domain_id:
        logger.warning(
            'Sub-domain %s does not belong to domain %s', sub_domain, statistic.group.domain
        )
        return

    rows = [
        StatisticValue(
            statistic=statistic, sub_domain=sub_domain, reference=reference, value=Decimal(1)
        )
        for reference in references
    ]

    with transaction.atomic():
        StatisticValue.objects.bulk_create(rows, batch_size=500)

track_configured classmethod

Source code in django_spire/metric/domain/statistic/services/tracking_service.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
@classmethod
def track_configured(cls, *, reference: str = 'page_click') -> StatisticValue | None:
    from django_spire.metric.domain.models import SubDomain  # noqa: PLC0415
    from django_spire.metric.domain.statistic.models import Statistic  # noqa: PLC0415

    statistic_key = settings.DJANGO_SPIRE_INTERNAL_METRIC_STATISTIC_KEY
    sub_domain_key = settings.DJANGO_SPIRE_INTERNAL_METRIC_SUB_DOMAIN_KEY

    if (not statistic_key) or (not sub_domain_key):
        return None

    statistic = Statistic.objects.for_key(statistic_key).active().not_deleted().first()
    sub_domain = SubDomain.objects.for_key(sub_domain_key).active().not_deleted().first()

    if statistic is None or sub_domain is None:
        logger.debug('Statistic tracking target not found')
        return None

    return statistic.services.tracking.track(sub_domain, reference=reference)

prune_retention classmethod

Source code in django_spire/metric/domain/statistic/services/tracking_service.py 
123
124
125
126
127
128
129
130
131
132
133
134
@classmethod
def prune_retention(cls, *, retention_days: int | None = None) -> int:
    from django_spire.metric.domain.statistic.models import StatisticValue  # noqa: PLC0415

    if retention_days is None:
        retention_days = getattr(settings, 'DJANGO_SPIRE_METRIC_RETENTION_DAYS', 90)

    if retention_days <= 0:
        return 0

    cutoff = timezone.now() - timedelta(days=retention_days)
    return cls._delete_batch(StatisticValue.objects.filter(timestamp__lt=cutoff), [])

trim_all classmethod

Source code in django_spire/metric/domain/statistic/services/tracking_service.py 
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
@classmethod
def trim_all(cls, *, max_values: int | None = None) -> int:
    from django_spire.metric.domain.statistic.models import StatisticValue  # noqa: PLC0415

    cap = max_values or getattr(settings, 'DJANGO_SPIRE_METRIC_TRACKING_VALUES_MAX', 1000)

    groups = (
        StatisticValue.objects.values('statistic_id', 'sub_domain_id', 'reference')
        .annotate(count=Count('pk'))
        .filter(count__gt=cap)
    )

    total = 0
    for group in groups.iterator():
        values = StatisticValue.objects.filter(
            statistic_id=group['statistic_id'],
            sub_domain_id=group['sub_domain_id'],
            reference=group['reference'],
        )
        retained_pks = list(values.order_by('-timestamp').values_list('pk', flat=True)[:cap])
        total += cls._delete_batch(values, retained_pks)

    return total