Report.aggregate: param data -> iter get_data

kiwi_simple_metrics/metrics/_report.py

@@ -1,5 +1,5 @@
-from dataclasses import dataclass, field
-from typing import Self
+from dataclasses import dataclass
+from typing import Callable, Iterator, Self
 
 from ..settings import MetricSettings
 
@@ -17,35 +17,39 @@ class ReportData:
         )
 
     def report(self, settings: MetricSettings) -> "Report":
-        result = settings.report.format(
+        return Report(
+            result=settings.report.format(
                 name=self.name,
                 value=self.value,
-        )
-
-        return Report(result, failed=(
+            ),
+            failed=(
                 self.value > settings.threshold and not settings.inverted
                 or self.value < settings.threshold and settings.inverted
-        ))
+            ),
+        )
 
 
-@dataclass(slots=True, frozen=True)
+@dataclass(slots=True, kw_only=True, frozen=True)
 class Report:
     result: str
-    failed: bool = field(default=False, kw_only=True)
+    failed: bool = False
 
     @classmethod
     def aggregate(
         cls, *,
         settings: MetricSettings,
-        data: list[ReportData],
-    ) -> Self:
+        get_data: Callable[[], Iterator[ReportData]],
+    ) -> Self | None:
+        if not settings.enabled:
+            return None
+
         reports = [
             data.report(settings)
-            for data in data
+            for data in get_data()
         ]
 
         return cls(
-            settings.report_outer.format(
+            result=settings.report_outer.format(
                 name=settings.name,
                 inner=", ".join(
                     report.result

kiwi_simple_metrics/metrics/cpu.py

@@ -1,20 +1,20 @@
+from typing import Iterator
+
 import psutil
 
 from ..settings import SETTINGS
 from ._report import Report, ReportData
 
 
-def _hwdata() -> ReportData:
-    return ReportData(
+def _hwdata() -> Iterator[ReportData]:
+    yield ReportData(
         name=SETTINGS.cpu.name,
         value=psutil.cpu_percent(interval=1),
     )
 
 
 def cpu() -> Report | None:
-    if not SETTINGS.cpu.enabled:
-        return None
-
-    data = _hwdata()
-
-    return data.report(SETTINGS.cpu)
+    return Report.aggregate(
+        settings=SETTINGS.cpu,
+        get_data=_hwdata,
+    )

kiwi_simple_metrics/metrics/disk.py

@@ -1,10 +1,11 @@
 import os
+from typing import Iterator
 
 from ..settings import SETTINGS
 from ._report import Report, ReportData
 
 
-def _hwdata() -> list[ReportData]:
+def _hwdata() -> Iterator[ReportData]:
     def get_path_statvfs(path: os.PathLike) -> dict[str, float]:
         sv = os.statvfs(path)
         return {
@@ -12,7 +13,7 @@ def _hwdata() -> list[ReportData]:
             "total": sv.f_blocks,
         }
 
-    return sorted([
+    yield from sorted([
         ReportData.from_free_total(
             name=str(path),
             **get_path_statvfs(path),
@@ -21,12 +22,7 @@ def _hwdata() -> list[ReportData]:
 
 
 def disk() -> Report | None:
-    if not SETTINGS.disk.enabled:
-        return None
-
-    data = _hwdata()
-
     return Report.aggregate(
         settings=SETTINGS.disk,
-        data=data,
+        get_data=_hwdata,
     )

kiwi_simple_metrics/metrics/memory.py

@@ -1,43 +1,41 @@
+from typing import Iterator
+
 import psutil
 
 from ..settings import SETTINGS
 from ._report import Report, ReportData
 
 
-def _hwdata() -> list[ReportData]:
+def _hwdata() -> Iterator[ReportData]:
     vmem = psutil.virtual_memory()
     swap = psutil.swap_memory()
 
     if SETTINGS.memory.swap == "exclude":
-        return [ReportData(
+        yield ReportData(
             name=SETTINGS.memory.name_ram,
             value=vmem.percent,
-        )]
+        )
 
     elif SETTINGS.memory.swap == "combine":
-        return [ReportData.from_free_total(
+        yield ReportData.from_free_total(
             name=SETTINGS.memory.name,
             free=vmem.available + swap.free,
             total=vmem.total + swap.total,
-        )]
+        )
 
     else:  # SETTINGS.memory.swap == "include"
-        return [ReportData(
+        yield ReportData(
             name=SETTINGS.memory.name_ram,
             value=vmem.percent,
-        ), ReportData(
+        )
+        yield ReportData(
             name=SETTINGS.memory.name_swap,
             value=swap.percent,
-        )]
+        )
 
 
 def memory() -> Report | None:
-    if not SETTINGS.memory.enabled:
-        return None
-
-    data = _hwdata()
-
     return Report.aggregate(
         settings=SETTINGS.memory,
-        data=data,
+        get_data=_hwdata,
     )

kiwi_simple_metrics/settings.py

@@ -22,7 +22,7 @@ class MetricSettings(BaseModel):
     report: str = "{name}: {value:.2f}%"
 
     # per-metric format string for reporting
-    report_outer: str = "{name}: [{inner}]"
+    report_outer: str = "{inner}"
 
     # include only `count` many items (None: include all)
     count: int | None = None
@@ -36,7 +36,6 @@ class CpuMS(MetricSettings):
 class MemoryMS(MetricSettings):
     name: str = "Memory"
     threshold: float = 90
-    report_outer: str = "{inner}"
 
     # how to handle swap space
     # exclude: swap space is not reported
@@ -52,6 +51,7 @@ class MemoryMS(MetricSettings):
 class DiskMS(MetricSettings):
     name: str = "Disk Used"
     threshold: float = 85
+    report_outer: str = "{name}: [{inner}]"
     count: int = 1
 
     # paths to check for disk space