Coverage for aiopromql/models/core.py: 91%

1""" 

2Generic data structures for modeling time series and labeled metrics. 

3""" 

4 

5from datetime import datetime 

6from typing import Dict, List, NamedTuple 

7 

8 

9class MetricLabelSet: 

10 """ 

11 Hashable wrapper around a Prometheus metric label dictionary. 

12 

13 Prometheus metrics are identified by a set of key-value labels 

14 (e.g., {"job": "api", "instance": "localhost:9090"}). This class allows such 

15 a label set to be used as a key in Python dictionaries by making it hashable 

16 and comparable. 

17 

18 Instances of this class are used as keys in the dictionary returned by 

19 `PrometheusResponseModel.to_metric_map()`, where each MetricLabelSet maps to 

20 a TimeSeries object. 

21 """ 

22 

23 def __init__(self, metric: Dict[str, str]): 

24 self.dict = metric 

25 self._key = frozenset(metric.items()) 

26 

27 def __hash__(self) -> int: 

28 return hash(self._key) 

29 

30 def __eq__(self, other) -> bool: 

31 if not isinstance(other, MetricLabelSet): 

32 return False 

33 return self._key == other._key 

34 

35 def __repr__(self) -> str: 

36 return f"MetricLabelSet({self.dict})" 

37 

38 def get(self, label: str, default=None): 

39 """ 

40 Return the value for the given label key, or default if not present. 

41 

42 :param label: The label key to retrieve from the metric dictionary. 

43 :type label: str 

44 :param default: The value to return if the label is not found. Defaults to None. 

45 :return: The value corresponding to the label, or the default if label is missing. 

46 :rtype: str or Any 

47 """ 

48 return self.dict.get(label, default) 

49 

50 

51class TimeSeriesPoint(NamedTuple): 

52 """ 

53 A single timestamped data point from a Prometheus time series. 

54 

55 Represents one (timestamp, value) pair, where the timestamp is a `datetime` 

56 object and the value is a float. Useful for building time series from Prometheus 

57 query results. 

58 """ 

59 

60 timestamp: datetime 

61 value: float 

62 

63 @classmethod 

64 def from_prometheus_value(cls, ts: float, value: str) -> "TimeSeriesPoint": 

65 """ 

66 Converts a Prometheus response (timestamp, value) pair to TimeSeriesPoint. 

67 

68 Args: 

69 ts: Epoch timestamp. 

70 value: String representation of float value. 

71 

72 Returns: 

73 A TimeSeriesPoint instance. 

74 """ 

75 return cls(datetime.fromtimestamp(ts), float(value)) 

76 

77 def __str__(self): 

78 return f"{self.timestamp.isoformat()} → {self.value:.2f}" 

79 

80 

81class TimeSeries: 

82 """ 

83 A sequence of timestamped float values (TimeSeriesPoint) with utility methods. 

84 

85 This class abstracts a Prometheus time series and provides methods for inspection, 

86 aggregation, and manipulation. Used in `PrometheusResponseModel.to_metric_map()` 

87 where each MetricLabelSet maps to a TimeSeries. 

88 """ 

89 

90 def __init__(self, values: List[TimeSeriesPoint]): 

91 """ 

92 Args: 

93 values: List of initial TimeSeriesPoint objects. 

94 """ 

95 self.values: List[TimeSeriesPoint] = values 

96 

97 def __iter__(self): 

98 return iter(self.values) 

99 

100 def __len__(self): 

101 return len(self.values) 

102 

103 def __getitem__(self, idx) -> TimeSeriesPoint: 

104 return self.values[idx] 

105 

106 def __repr__(self): 

107 return f"Values({self.values})" 

108 

109 def add_point(self, point: TimeSeriesPoint): 

110 """Adds a new data point.""" 

111 self.values.append(point) 

112 

113 def extend(self, other: "TimeSeries"): 

114 """Appends another TimeSeries' points to this one.""" 

115 self.values.extend(other.values) 

116 

117 def latest(self) -> TimeSeriesPoint | None: 

118 """Returns the latest (most recent) data point.""" 

119 return max(self.values, key=lambda x: x.timestamp, default=None) 

120 

121 def average(self) -> float | None: 

122 """Computes the average of all values.""" 

123 nums = [v.value for v in self.values if isinstance(v.value, (int, float))] 

124 return sum(nums) / len(nums) if nums else None