mirror of
https://github.com/yavook/kiwi-simple-metrics.git
synced 2024-11-22 15:42:59 +00:00
121 lines
2.9 KiB
Python
121 lines
2.9 KiB
Python
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,
|
|
)
|