or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

docs

advanced.mdcollectors.mdcontext-managers.mdcore-metrics.mdexposition.mdindex.mdregistry.md
tile.json

tessl/pypi-prometheus-client

Python client for the Prometheus monitoring system.

Workspace
tessl
Visibility
Public
Created
Last updated
Describes
pypipkg:pypi/prometheus-client@0.22.x

To install, run

npx @tessl/cli install tessl/pypi-prometheus-client@0.22.0

index.mddocs/

Prometheus Client

The 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.

Package Information

  • Package Name: prometheus-client
  • Language: Python
  • Installation: pip install prometheus-client

Core Imports

import prometheus_client

Most common imports for metrics:

from prometheus_client import Counter, Gauge, Histogram, Summary, Info, Enum

Registry and exposition:

from prometheus_client import REGISTRY, generate_latest, start_http_server

Basic Usage

from 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)

Architecture

The prometheus-client library is built around several key components:

  • Metrics: Core measurement types (Counter, Gauge, Histogram, Summary, Info, Enum) that collect data
  • Registry: Central collection point that manages all metrics and their metadata
  • Collectors: Components that generate metrics data, including built-in system collectors
  • Exposition: Functions and servers that output metrics in Prometheus format for scraping
  • Context Managers: Utilities for automatic instrumentation of timing, exceptions, and progress tracking

This design provides thread-safe metric collection, flexible labeling for multi-dimensional data, and multiple exposition methods to integrate with various deployment architectures.

Utility Functions

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 needed

Capabilities

Core Metrics

The 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: ...

Core Metrics

Registry and Collection

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 registry

Registry System

Exposition and Servers

Functions 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: ...

Exposition

Built-in Collectors

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 collector

Built-in Collectors

Context Managers and Utilities

Context 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 usage

Context Managers

Custom Collectors and Advanced Features

Advanced 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: ...

Advanced Features

Types

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: ...