from __future__ import annotations import abc import dataclasses import time import typing from ._path import ( Labels, LabelsData, ) class Now: __slots__ = () def time(self) -> int: """Time since epoch in milliseconds""" return round(time.time() * 1000) NOW = Now() DynamicTimestamp = int | Now | None def _to_timestamp(timestamp: DynamicTimestamp) -> int | None: if isinstance(timestamp, Now): return timestamp.time() else: return timestamp @dataclasses.dataclass(slots=True, kw_only=True) class MetricValueSet: # list of suffixes and values; suffixes must match the allowed suffixes of the family of the point values: list[tuple[str, float]] timestamp: int | None _TMetricPoint = typing.TypeVar('_TMetricPoint', bound='MetricPoint') class MetricPoint(abc.ABC): # _family and _labels set by MetricFamily._init_point # __weakref__ is needed for weakref support __slots__ = ('_family', '_labels', '__weakref__') _family: MetricFamily _labels: Labels @property def family(self) -> MetricFamily: return self._family @property def labels(self) -> Labels: return self._labels def __repr__(self) -> str: return f"" @abc.abstractmethod def _get_value_set(self) -> MetricValueSet: ... class MetricFamily(abc.ABC, typing.Generic[_TMetricPoint]): TYPE: typing.ClassVar[str] SUFFIXES: typing.ClassVar[list[str]] = [''] RESERVED_LABELS: typing.ClassVar[list[str]] = [] def __init__( self, *, name: str, unit: str = "", help: str = "", ) -> None: self.name = name self.unit = unit self.help = help def _init_point(self, labels: LabelsData, point: _TMetricPoint) -> _TMetricPoint: if not isinstance(labels, Labels): labels = Labels(labels) for reserved_label in self.RESERVED_LABELS: if reserved_label in labels.map: raise KeyError(f"Point {labels.key} uses reserved label {reserved_label}") point._family = self point._labels = labels return point