Python client for the Prometheus monitoring system.
npx @tessl/cli install tessl/pypi-prometheus-client@0.22.0The official Python client library for Prometheus, providing comprehensive instrumentation capabilities for collecting and exposing metrics. This library enables applications to define, collect, and export metrics in Prometheus format, supporting counters, gauges, histograms, summaries, and specialized metric types for monitoring application performance and system health.
pip install prometheus-clientimport prometheus_clientMost common imports for metrics:
from prometheus_client import Counter, Gauge, Histogram, Summary, Info, EnumRegistry and exposition:
from prometheus_client import REGISTRY, generate_latest, start_http_serverfrom prometheus_client import Counter, Gauge, start_http_server
import time
import random
# Create metrics
request_count = Counter('http_requests_total', 'Total HTTP requests', ['method', 'status'])
response_time = Gauge('http_response_time_seconds', 'HTTP response time')
# Use metrics
request_count.labels(method='GET', status='200').inc()
response_time.set(0.25)
# Start metrics server
start_http_server(8000)
# Keep the application running
while True:
# Simulate some work
request_count.labels(method='GET', status='200').inc()
response_time.set(random.uniform(0.1, 0.5))
time.sleep(1)The prometheus-client library is built around several key components:
This design provides thread-safe metric collection, flexible labeling for multi-dimensional data, and multiple exposition methods to integrate with various deployment architectures.
Global configuration and utility functions for metric management.
def enable_created_metrics() -> None:
"""Enable exporting _created metrics on counters, histograms, and summaries."""
def disable_created_metrics() -> None:
"""Disable exporting _created metrics on counters, histograms, and summaries."""
def floatToGoString(d: float) -> str:
"""Convert float to Go-style string representation."""
def instance_ip_grouping_key() -> Dict[str, str]:
"""Return grouping key with instance IP for push gateway."""Usage Example:
from prometheus_client import enable_created_metrics, disable_created_metrics
# Control _created series export for better performance
disable_created_metrics() # Disable to save memory and bandwidth
# ... create metrics ...
enable_created_metrics() # Re-enable if neededThe fundamental metric types for collecting different kinds of measurements, including monotonic counters, up/down gauges, distribution histograms, statistical summaries, and key-value info metrics.
class Counter:
def __init__(self, name: str, documentation: str, labelnames=(), namespace='', subsystem='', unit='', registry=REGISTRY) -> None: ...
def inc(self, amount: float = 1, exemplar: Optional[Dict[str, str]] = None) -> None: ...
def labels(self, *labelvalues, **labelkwargs) -> Counter: ...
class Gauge:
def __init__(self, name: str, documentation: str, labelnames=(), namespace='', subsystem='', unit='', registry=REGISTRY, multiprocess_mode='all') -> None: ...
def set(self, value: float) -> None: ...
def inc(self, amount: float = 1) -> None: ...
def dec(self, amount: float = 1) -> None: ...
def labels(self, *labelvalues, **labelkwargs) -> Gauge: ...
class Histogram:
def __init__(self, name: str, documentation: str, labelnames=(), namespace='', subsystem='', unit='', registry=REGISTRY, buckets=DEFAULT_BUCKETS) -> None: ...
def observe(self, amount: float, exemplar: Optional[Dict[str, str]] = None) -> None: ...
def labels(self, *labelvalues, **labelkwargs) -> Histogram: ...
class Summary:
def __init__(self, name: str, documentation: str, labelnames=(), namespace='', subsystem='', unit='', registry=REGISTRY) -> None: ...
def observe(self, amount: float) -> None: ...
def labels(self, *labelvalues, **labelkwargs) -> Summary: ...
class Info:
def __init__(self, name: str, documentation: str, labelnames=(), namespace='', subsystem='', registry=REGISTRY) -> None: ...
def info(self, val: Dict[str, str]) -> None: ...
class Enum:
def __init__(self, name: str, documentation: str, labelnames=(), namespace='', subsystem='', registry=REGISTRY, states=None) -> None: ...
def state(self, state: str) -> None: ...The registry system that manages metric collection, organization, and retrieval, providing centralized access to all metrics and enabling custom collectors.
class CollectorRegistry:
def __init__(self, auto_describe: bool = False, target_info: Optional[Dict[str, str]] = None) -> None: ...
def register(self, collector: Collector) -> None: ...
def unregister(self, collector: Collector) -> None: ...
def collect(self) -> Iterable[Metric]: ...
def get_sample_value(self, name: str, labels: Optional[Dict[str, str]] = None) -> Optional[float]: ...
REGISTRY: CollectorRegistry # Global default registryFunctions and servers for outputting metrics in Prometheus format, including HTTP servers, WSGI/ASGI applications, file output, and push gateway integration.
def generate_latest(registry: CollectorRegistry = REGISTRY) -> bytes: ...
def start_http_server(port: int, addr: str = '0.0.0.0', registry: CollectorRegistry = REGISTRY) -> Tuple[WSGIServer, threading.Thread]: ...
def make_wsgi_app(registry: CollectorRegistry = REGISTRY, disable_compression: bool = False) -> Callable: ...
def make_asgi_app(registry: CollectorRegistry = REGISTRY, disable_compression: bool = False) -> Callable: ...
def write_to_textfile(path: str, registry: CollectorRegistry) -> None: ...
def push_to_gateway(gateway: str, job: str, registry: CollectorRegistry, grouping_key=None, timeout: Optional[float] = 30) -> None: ...
def pushadd_to_gateway(gateway: str, job: str, registry: Optional[CollectorRegistry], grouping_key=None, timeout: Optional[float] = 30) -> None: ...
def delete_from_gateway(gateway: str, job: str, grouping_key=None, timeout: Optional[float] = 30) -> None: ...Pre-built collectors for system metrics including process statistics, garbage collection data, and platform information that can be automatically registered for comprehensive monitoring.
class ProcessCollector:
def __init__(self, namespace: str = '', pid: Callable = lambda: 'self', proc: str = '/proc', registry: Optional[CollectorRegistry] = REGISTRY) -> None: ...
class GCCollector:
def __init__(self, registry: CollectorRegistry = REGISTRY) -> None: ...
class PlatformCollector:
def __init__(self, registry: Optional[CollectorRegistry] = REGISTRY, platform=None) -> None: ...
PROCESS_COLLECTOR: ProcessCollector # Default process collector
GC_COLLECTOR: GCCollector # Default GC collector
PLATFORM_COLLECTOR: PlatformCollector # Default platform collectorContext managers and decorators for automatic instrumentation including timing operations, counting exceptions, and tracking in-progress work.
class Timer:
def __enter__(self): ...
def __exit__(self, typ, value, traceback): ...
def __call__(self, f) -> Callable: ... # Decorator usage
class ExceptionCounter:
def __enter__(self) -> None: ...
def __exit__(self, typ, value, traceback) -> bool: ...
def __call__(self, f) -> Callable: ... # Decorator usage
class InprogressTracker:
def __enter__(self): ...
def __exit__(self, typ, value, traceback): ...
def __call__(self, f) -> Callable: ... # Decorator usageAdvanced functionality for building custom metric collectors, multiprocess support, and specialized integrations including metric families, sample types, and bridge components.
class Metric:
def __init__(self, name: str, documentation: str, typ: str, unit: str = '') -> None: ...
def add_sample(self, name: str, labels: Dict[str, str], value: float, timestamp=None, exemplar=None) -> None: ...
class Collector:
def collect(self) -> Iterable[Metric]: ... # Abstract method
class MultiProcessCollector:
def __init__(self, registry, path=None) -> None: ...from typing import Dict, Optional, Iterable, Callable, Union, Sequence, NamedTuple, Literal
# Sample and data types
class Sample(NamedTuple):
name: str
labels: Dict[str, str]
value: float
timestamp: Optional[Union[float, Timestamp]] = None
exemplar: Optional[Exemplar] = None
native_histogram: Optional[NativeHistogram] = None
class Exemplar(NamedTuple):
labels: Dict[str, str]
value: float
timestamp: Optional[Union[float, Timestamp]] = None
class Timestamp:
def __init__(self, sec: float, nsec: float) -> None: ...
def __float__(self) -> float: ...
class BucketSpan(NamedTuple):
offset: int
length: int
class NativeHistogram(NamedTuple):
schema: int
zero_threshold: float
zero_count: int
count: int
sum: float
positive_spans: Sequence[BucketSpan]
positive_deltas: Sequence[int]
negative_spans: Sequence[BucketSpan]
negative_deltas: Sequence[int]
# Metric metadata
class Metric:
def __init__(self, name: str, documentation: str, typ: str, unit: str = '') -> None: ...
def add_sample(self, name: str, labels: Dict[str, str], value: float,
timestamp: Optional[Union[float, Timestamp]] = None,
exemplar: Optional[Exemplar] = None,
native_histogram: Optional[NativeHistogram] = None) -> None: ...
# Base collector interface
class Collector:
def collect(self) -> Iterable[Metric]: ...
def describe(self) -> Iterable[Metric]: ...
# Registry
class CollectorRegistry:
def __init__(self, auto_describe: bool = False, target_info: Optional[Dict[str, str]] = None) -> None: ...
def register(self, collector: Collector) -> None: ...
def unregister(self, collector: Collector) -> None: ...
def collect(self) -> Iterable[Metric]: ...
def get_sample_value(self, name: str, labels: Optional[Dict[str, str]] = None) -> Optional[float]: ...
# Constants
INF: float = float("inf")
MINUS_INF: float = float("-inf")
NaN: float = float("NaN")
CONTENT_TYPE_LATEST: str = 'text/plain; version=0.0.4; charset=utf-8'
# Global registry instance
REGISTRY: CollectorRegistry
# Validation and utility functions
def enable_created_metrics() -> None: ...
def disable_created_metrics() -> None: ...
def floatToGoString(d: float) -> str: ...