processor_service

django_spire.metric.domain.statistic.services.processor_service

StatisticGroupProcessorService

Bases: BaseDjangoModelService['StatisticGroup']

Source code in django_spire/contrib/constructor/constructor.py

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

StatisticProcessorService

Bases: BaseDjangoModelService['Statistic']

Source code in django_spire/contrib/constructor/constructor.py

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

add_value

Source code in django_spire/metric/domain/statistic/services/processor_service.py

def add_value(
    self,
    reference: str,
    sub_domain: SubDomain,
    value: float | str | Decimal = 1,
    *,
    value_timestamp: datetime | None = None,
) -> StatisticValue:
    if sub_domain.domain_id != self.obj.group.domain_id:
        message = (
            f"Sub-domain '{sub_domain}' does not belong to domain "
            f"'{self.obj.group.domain}'"
        )
        raise ServiceError(message)

    value = Decimal(value)
    stamp = value_timestamp or timezone.now()
    if timezone.is_naive(stamp):
        stamp = timezone.make_aware(stamp)

    statistic_value = self.obj.values.create(
        sub_domain=sub_domain, reference=reference, timestamp=stamp, value=value
    )
    statistic_value.value = statistic_value.value.quantize(self._value_precision())

    return statistic_value

subtract_value

Source code in django_spire/metric/domain/statistic/services/processor_service.py

def subtract_value(
    self,
    reference: str,
    sub_domain: SubDomain,
    value: float | str | Decimal = 1,
    *,
    value_timestamp: datetime | None = None,
) -> StatisticValue:
    return self.add_value(
        reference, sub_domain, -Decimal(value), value_timestamp=value_timestamp
    )

increment

Source code in django_spire/metric/domain/statistic/services/processor_service.py

def increment(
    self, reference: str, sub_domain: SubDomain, *, value_timestamp: datetime | None = None
) -> StatisticValue:
    return self.add_value(reference, sub_domain, Decimal(1), value_timestamp=value_timestamp)

decrement

Source code in django_spire/metric/domain/statistic/services/processor_service.py

def decrement(
    self, reference: str, sub_domain: SubDomain, *, value_timestamp: datetime | None = None
) -> StatisticValue:
    return self.add_value(reference, sub_domain, Decimal(-1), value_timestamp=value_timestamp)

StatisticValueProcessorService

Bases: BaseDjangoModelService['StatisticValue']

Source code in django_spire/contrib/constructor/constructor.py

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