Skip to content

service

django_spire.metric.domain.statistic.services.service

logger = logging.getLogger(__name__) module-attribute

StatisticService

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

intelligence = StatisticIntelligenceService() class-attribute instance-attribute

processor = StatisticProcessorService() class-attribute instance-attribute

factory = StatisticFactoryService() class-attribute instance-attribute

tracking = StatisticTrackingService() class-attribute instance-attribute

transformation = StatisticTransformationService() class-attribute instance-attribute

record classmethod

Source code in django_spire/metric/domain/statistic/services/service.py
@classmethod
def record(
    cls,
    statistic_key: str,
    sub_domain_key: str,
    reference: str,
    value: float | str | Decimal = 1,
) -> StatisticValue:
    from django_spire.metric.domain.models import SubDomain  # noqa: PLC0415
    from django_spire.metric.domain.statistic.models import (  # noqa: PLC0415
        Statistic,
        StatisticValue,
    )

    reference_max_length = StatisticValue._meta.get_field('reference').max_length
    if len(reference) > reference_max_length:
        message = (
            f"Reference '{reference}' exceeds the maximum length "
            f'of {reference_max_length} characters'
        )
        raise ServiceError(message)

    statistic = Statistic.objects.for_key(statistic_key).active().not_deleted()
    statistic_ids = Subquery(statistic.values('pk')[:1])
    sub_domain_ids = Subquery(
        SubDomain.objects.for_key(sub_domain_key)
        .filter(domain_id__in=Subquery(statistic.values('group__domain_id')[:1]))
        .active()
        .not_deleted()
        .values('pk')[:1]
    )

    value_precision = StatisticValue._meta.get_field('value').decimal_places

    try:
        statistic_value = StatisticValue.objects.create(
            statistic_id=statistic_ids,
            sub_domain_id=sub_domain_ids,
            reference=reference,
            value=Decimal(value).quantize(Decimal(1).scaleb(-value_precision)),
        )
    except IntegrityError as error:
        message = f"Statistic '{statistic_key}' or sub-domain '{sub_domain_key}' not found"
        raise ServiceError(message) from error

    statistic_value.refresh_from_db()
    return statistic_value

remote_record classmethod

Source code in django_spire/metric/domain/statistic/services/service.py
@classmethod
def remote_record(
    cls,
    statistic_key: str,
    sub_domain_key: str,
    reference: str,
    value: float | str | Decimal = 1,
) -> dict | None:
    base_url = settings.DJANGO_SPIRE_REMOTE_API_URL
    key = settings.DJANGO_SPIRE_REMOTE_API_KEY

    if not base_url or not key:
        logger.warning(
            'Remote metric record skipped: DJANGO_SPIRE_REMOTE_API_URL or '
            'DJANGO_SPIRE_REMOTE_API_KEY is not configured'
        )
        return None

    from django_spire.metric.domain.statistic.services.rest_connector import (  # noqa: PLC0415
        SpireMetricRestConnector,
    )

    payload = {
        'reference': reference,
        'sub_domain_key': sub_domain_key,
        'value': str(Decimal(value)),
    }

    try:
        connector = SpireMetricRestConnector(base_url=base_url, api_key=key)
        return connector.record(statistic_key, payload)
    except Exception:
        logger.exception('Remote metric record failed')
        return None

StatisticGroupService

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

intelligence = StatisticGroupIntelligenceService() class-attribute instance-attribute

processor = StatisticGroupProcessorService() class-attribute instance-attribute

factory = StatisticGroupFactoryService() class-attribute instance-attribute

transformation = StatisticGroupTransformationService() class-attribute instance-attribute

StatisticValueService

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

intelligence = StatisticValueIntelligenceService() class-attribute instance-attribute

processor = StatisticValueProcessorService() class-attribute instance-attribute

factory = StatisticValueFactoryService() class-attribute instance-attribute

transformation = StatisticValueTransformationService() class-attribute instance-attribute