File Manager

✏️ t/registry.py

import copy
from threading import Lock
from typing import Dict, Iterable, List, Optional, Protocol, Set

from .metrics_core import Metric

class Collector(Protocol):
    def collect(self) -> Iterable[Metric]:
        """Collect metrics."""

class _EmptyCollector:
    def collect(self) -> Iterable[Metric]:
        return []

class DuplicateTimeseries(ValueError):
    def __init__(self, duplicates: Set[str]):
        msg = 'Duplicated timeseries in CollectorRegistry: {}'.format(
            duplicates)
        super().__init__(msg)
        self.duplicates: Set[str] = duplicates

class CollectorRegistry:
    """Metric collector registry.

Collectors must have a no-argument method 'collect' that returns a list of
    Metric objects. The returned metrics should be consistent with the Prometheus
    exposition formats.
    """

def __init__(self, auto_describe: bool = False, target_info: Optional[Dict[str, str]] = None,
                 support_collectors_without_names: bool = False):
        self._collector_to_names: Dict[Collector, List[str]] = {}
        self._names_to_collectors: Dict[str, Collector] = {}
        self._auto_describe = auto_describe
        self._lock = Lock()
        self._target_info: Optional[Dict[str, str]] = {}
        self._support_collectors_without_names = support_collectors_without_names
        self._collectors_without_names: List[Collector] = []
        self.set_target_info(target_info)

def register(self, collector: Collector) -> None:
        """Add a collector to the registry."""
        with self._lock:
            names = self._get_names(collector)
            duplicates = set(self._names_to_collectors).intersection(names)
            if duplicates:
                raise DuplicateTimeseries(duplicates)
            for name in names:
                self._names_to_collectors[name] = collector
            self._collector_to_names[collector] = names
            if self._support_collectors_without_names and not names:
                self._collectors_without_names.append(collector)

def unregister(self, collector: Collector) -> None:
        """Remove a collector from the registry."""
        with self._lock:
            for name in self._collector_to_names[collector]:
                del self._names_to_collectors[name]
            del self._collector_to_names[collector]
            if collector in self._collectors_without_names:
                self._collectors_without_names.remove(collector)

def _get_names(self, collector):
        """Get names of timeseries the collector produces and clashes with."""
        desc_func = None
        # If there's a describe function, use it.
        try:
            desc_func = collector.describe
        except AttributeError:
            pass
        # Otherwise, if auto describe is enabled use the collect function.
        if not desc_func and self._auto_describe:
            desc_func = collector.collect

if not desc_func:
            return []

result = []
        type_suffixes = {
            'counter': ['_total', '_created'],
            'summary': ['_sum', '_count', '_created'],
            'histogram': ['_bucket', '_sum', '_count', '_created'],
            'gaugehistogram': ['_bucket', '_gsum', '_gcount'],
            'info': ['_info'],
        }
        for metric in desc_func():
            result.append(metric.name)
            for suffix in type_suffixes.get(metric.type, []):
                result.append(metric.name + suffix)
        return result

def collect(self) -> Iterable[Metric]:
        """Yields metrics from the collectors in the registry."""
        collectors = None
        ti = None
        with self._lock:
            collectors = copy.copy(self._collector_to_names)
            if self._target_info:
                ti = self._target_info_metric()
        if ti:
            yield ti
        for collector in collectors:
            yield from collector.collect()

def restricted_registry(self, names: Iterable[str]) -> "RestrictedRegistry":
        """Returns object that only collects some metrics.

Returns an object which upon collect() will return
        only samples with the given names.

Intended usage is:
            generate_latest(REGISTRY.restricted_registry(['a_timeseries']), escaping)

Experimental."""
        names = set(names)
        return RestrictedRegistry(names, self)

def set_target_info(self, labels: Optional[Dict[str, str]]) -> None:
        with self._lock:
            if labels:
                if not self._target_info and 'target_info' in self._names_to_collectors:
                    raise ValueError('CollectorRegistry already contains a target_info metric')
                self._names_to_collectors['target_info'] = _EmptyCollector()
            elif self._target_info:
                self._names_to_collectors.pop('target_info', None)
            self._target_info = labels

def get_target_info(self) -> Optional[Dict[str, str]]:
        with self._lock:
            return self._target_info

def _target_info_metric(self):
        m = Metric('target', 'Target metadata', 'info')
        m.add_sample('target_info', self._target_info, 1)
        return m

def get_sample_value(self, name: str, labels: Optional[Dict[str, str]] = None) -> Optional[float]:
        """Returns the sample value, or None if not found.

This is inefficient, and intended only for use in unittests.
        """
        if labels is None:
            labels = {}
        for metric in self.collect():
            for s in metric.samples:
                if s.name == name and s.labels == labels:
                    return s.value
        return None

class RestrictedRegistry:
    def __init__(self, names: Iterable[str], registry: CollectorRegistry):
        self._name_set = set(names)
        self._registry = registry

def collect(self) -> Iterable[Metric]:
        collectors = set(self._registry._collectors_without_names)
        target_info_metric = None
        with self._registry._lock:
            if 'target_info' in self._name_set and self._registry._target_info:
                target_info_metric = self._registry._target_info_metric()
            for name in self._name_set:
                if name != 'target_info' and name in self._registry._names_to_collectors:
                    collectors.add(self._registry._names_to_collectors[name])
        if target_info_metric:
            yield target_info_metric
        for collector in collectors:
            for metric in collector.collect():
                m = metric._restricted_metric(self._name_set)
                if m:
                    yield m

REGISTRY = CollectorRegistry(auto_describe=True)

Batal

⬅ site-packages 26 item

Nama	Tipe	Ukuran	Diubah	Aksi
📁__pycache__	dir	—	2026-07-29 13:48	📂 Buka
📁aiohttp	dir	—	2026-07-29 13:48	📂 Buka
📁bridge	dir	—	2026-07-29 13:48	📂 Buka
📁django	dir	—	2026-07-29 13:48	📂 Buka
📁openmetrics	dir	—	2026-07-29 13:48	📂 Buka
📁twisted	dir	—	2026-07-29 13:48	📂 Buka
🐍__init__.py	py	1.9 KB	2026-07-29 13:48	✏️ 👁️
🐍asgi.py	py	1.6 KB	2026-07-29 13:48	✏️ 👁️
🐍context_managers.py	py	2.3 KB	2026-07-29 13:48	✏️ 👁️
🐍core.py	py	978 B	2026-07-29 13:48	✏️ 👁️
🐍decorator.py	py	15.4 KB	2026-07-29 13:48	✏️ 👁️
🐍exposition.py	py	32.2 KB	2026-07-29 13:48	✏️ 👁️
🐍gc_collector.py	py	1.5 KB	2026-07-29 13:48	✏️ 👁️
🐍metrics.py	py	28.9 KB	2026-07-29 13:48	✏️ 👁️
🐍metrics_core.py	py	15.2 KB	2026-07-29 13:48	✏️ 👁️
🐍mmap_dict.py	py	5.3 KB	2026-07-29 13:48	✏️ 👁️
🐍multiprocess.py	py	7.9 KB	2026-07-29 13:48	✏️ 👁️
🐍parser.py	py	12.3 KB	2026-07-29 13:48	✏️ 👁️
🐍platform_collector.py	py	1.8 KB	2026-07-29 13:48	✏️ 👁️
🐍process_collector.py	py	3.8 KB	2026-07-29 13:48	✏️ 👁️
📄py.typed	typed	0 B	2026-07-29 13:48	✏️ 👁️
🐍registry.py	py	6.6 KB	2026-07-29 13:48	✏️ 👁️
🐍samples.py	py	2.3 KB	2026-07-29 13:48	✏️ 👁️
🐍utils.py	py	910 B	2026-07-29 13:48	✏️ 👁️
🐍validation.py	py	4.1 KB	2026-07-29 13:48	✏️ 👁️
🐍values.py	py	5.5 KB	2026-07-29 13:48	✏️ 👁️

✏️ Rename