kiwi-simple-metrics/kiwi_simple_metrics/metrics/_report.py

import urllib.parse
from dataclasses import dataclass
from typing import Any, Callable, Iterator, Self

import requests

from ..settings import SETTINGS, MetricSettings


@dataclass(slots=True, kw_only=True)
class ReportData:
    name: str
    value: float
    threshold: float
    inverted: bool
    format: str

    @classmethod
    def from_settings(
        cls,
        *,
        name: str,
        value: float,
        settings: MetricSettings,
    ) -> Self:
        return cls(
            name=name,
            value=value,
            threshold=settings.threshold,
            inverted=settings.inverted,
            format=settings.report,
        )

    @classmethod
    def from_free_total(
        cls,
        *,
        name: str,
        free: float,
        total: float,
        settings: MetricSettings,
    ) -> Self:
        return cls.from_settings(
            name=name,
            value=(total - free) / total * 100,
            settings=settings,
        )

    @property
    def report(self) -> "Report":
        return Report(
            result=self.format.format(
                name=self.name,
                value=self.value,
            ),
            failed=(
                self.value > self.threshold
                and not self.inverted
                or self.value < self.threshold
                and self.inverted
            ),
        )


@dataclass(slots=True, kw_only=True, frozen=True)
class Report:
    result: str
    failed: bool = False

    def __str__(self) -> str:
        state = "OK" if not self.failed else "FAIL"

        return SETTINGS.log.format.format(
            state=state,
            result=self.result,
        )

    @classmethod
    def summary(cls, *_reports: Any) -> Self:
        reports = [report for report in _reports if isinstance(report, Report)]

        return cls(
            result=SETTINGS.separator.join(report.result for report in reports),
            failed=any(report.failed for report in reports),
        )

    @classmethod
    def aggregate(
        cls,
        *,
        settings: MetricSettings,
        get_data: Callable[[], Iterator[ReportData]],
    ) -> Self | None:
        if not settings.enabled:
            return None

        reports = [data.report for data in get_data()]

        if not reports:
            return None

        return cls(
            result=settings.report_outer.format(
                name=settings.name,
                inner=SETTINGS.separator.join(
                    report.result for report in reports[: settings.count]
                ),
            ),
            failed=any(report.failed for report in reports),
        )

    def push_webhook(self) -> None:
        if (
            url := SETTINGS.webhook.url if not self.failed else SETTINGS.webhook.fail
        ) is None:
            return

        requests.get(
            url=str(url).format(result=urllib.parse.quote_plus(self.result)),
            verify=not SETTINGS.webhook.insecure,
        )