net/python-prometheus
3
0
Fork 0
Code Pull Requests Packages Releases Activity

fix small lints

Browse Source
This commit is contained in:
Stefan Bühler 2025-05-08 12:00:54 +02:00
parent a6b6ca9c8f
commit 65cb5d31f3
2 changed files with 19 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
src/prometheus_rus/_metrics.py
View File

@ -16,7 +16,12 @@ class Counter(MetricBase):
class Gauge(MetricBase): class Gauge(MetricBase):
metric_type: typing.ClassVar[str] = 'gauge' metric_type: typing.ClassVar[str] = 'gauge'
def __init__(self, value: float = float('nan'), path: typing.Optional[Path] = None, help: typing.Optional[str] = None): def __init__(
self,
value: float = float('nan'),
path: typing.Optional[Path] = None,
help: typing.Optional[str] = None,
):
super().__init__(value=value, help=help) super().__init__(value=value, help=help)
if not path is None: if not path is None:
GLOBAL_REGISTRY.register(path, self) GLOBAL_REGISTRY.register(path, self)

23
src/prometheus_rus/_registry.py
View File

@ -1,8 +1,6 @@
from threading import Lock from threading import Lock
import typing import typing
import weakref import weakref
import collections
import time
import math import math
from ._metric_base import MetricValue, MetricBase, MetricGroupDefinition, MetricGroupBase, EMPTY_GROUP_DEFINITION from ._metric_base import MetricValue, MetricBase, MetricGroupDefinition, MetricGroupBase, EMPTY_GROUP_DEFINITION
@ -49,10 +47,10 @@ class _MetricCollection:
if self.group_definition.reserved_labels != group_definition.reserved_labels: if self.group_definition.reserved_labels != group_definition.reserved_labels:
raise RuntimeError("reserved labels don't match for {}: {} != {}".format( raise RuntimeError("reserved labels don't match for {}: {} != {}".format(
path.name, self.group_definition.reserved_labels, group_definition.reserved_labels)) path.name, self.group_definition.reserved_labels, group_definition.reserved_labels))
for l in self.group_definition.reserved_labels: for label in self.group_definition.reserved_labels:
if l in path.labels.map: if label in path.labels.map:
raise RuntimeError("'{} {}' uses reserved label {}".format( raise RuntimeError("'{} {}' uses reserved label {}".format(
path.name, path.labels.key, l)) path.name, path.labels.key, label))
if path.labels.key in self.metrics: if path.labels.key in self.metrics:
raise RuntimeError("'{} {}' already registered".format( raise RuntimeError("'{} {}' already registered".format(
path.name, path.labels.key)) path.name, path.labels.key))
@ -66,8 +64,8 @@ class _MetricCollection:
for (add_path, m) in metric.metrics(): for (add_path, m) in metric.metrics():
assert ( assert (
add_path.name == '' or add_path.name in self.group_definition.reserved_suffixes) add_path.name == '' or add_path.name in self.group_definition.reserved_suffixes)
for l in add_path.labels.map: for label in add_path.labels.map:
assert l in self.group_definition.reserved_labels assert label in self.group_definition.reserved_labels
p = Path(path.name + add_path.name, {**path.labels.map, **add_path.labels.map}) p = Path(path.name + add_path.name, {**path.labels.map, **add_path.labels.map})
metrics[p] = m metrics[p] = m
g = None # set below, but need for cleanup g = None # set below, but need for cleanup
@ -155,7 +153,12 @@ class Registry:
del self._metrics[metric_name] del self._metrics[metric_name]
# requires lock # requires lock
def _get_collection(self, path: Path, metric_type: str, group_definition: typing.Optional[MetricGroupDefinition] = None): def _get_collection(
self,
path: Path,
metric_type: str,
group_definition: typing.Optional[MetricGroupDefinition] = None,
):
c = None c = None
try: try:
c = self._metrics[path.name] c = self._metrics[path.name]
@ -205,8 +208,8 @@ class Registry:
metrics = [] metrics = []
with self._lock: with self._lock:
for (metric_name, c) in self._metrics.items(): for (metric_name, c) in self._metrics.items():
l = [(k, m.get()) for (k, m) in c.metrics.items()] data = [(k, m.get()) for (k, m) in c.metrics.items()]
metrics.append((metric_name, c.metric_type, c.help_text, l)) metrics.append((metric_name, c.metric_type, c.help_text, data))
from io import StringIO from io import StringIO
result = StringIO() result = StringIO()