Pytheus

https://pythe.us
Admin
Pytheus is a modern Python library for collecting Prometheus metrics with built-in multiprocessing...
Tokens:11,208
Snippets:162
Trust Score:4.7
Update:6 months ago
Show doc for...
Tokens:
Raw
### Setup Project Environment

Source: https://pythe.us/quickstart

Commands to create a virtual environment and install dependencies.

```bash
# create the project directory
mkdir quickstart
cd quickstart

# create a python virtual environment and activate it
python3 -m venv venv
. venv/bin/activate

# install the required libraries for the quickstart
pip install pytheus
pip install flask
```

--------------------------------

### Install Pytheus via pip

Source: https://pythe.us/

Standard installation command for the library.

```bash
pip install pytheus
```

--------------------------------

### Example Logging Output

Source: https://pythe.us/rust_backend

Example log messages indicating the initialization of pipeline and backend threads. These logs help in monitoring the backend's status and diagnosing connection issues.

```text
INFO:pytheus_backend_rs:Starting pipeline thread....0
INFO:pytheus_backend_rs:Starting pipeline thread....1
INFO:pytheus_backend_rs:Starting pipeline thread....2
INFO:pytheus_backend_rs:Starting pipeline thread....3
INFO:pytheus_backend_rs:Starting BackendAction thread....
INFO:pytheus_backend_rs:RedisBackend initialized

```

--------------------------------

### Install Pytheus Rust backend

Source: https://pythe.us/

Installation command for the Rust-powered multiprocessing backend.

```bash
pip install pytheus-backend-rs
```

--------------------------------

### Install Pytheus with Redis backend

Source: https://pythe.us/

Installation commands for adding Redis support for multiprocessing.

```bash
pip install redis

# or everything in one command
pip install pytheus[redis]
```

--------------------------------

### Run the Flask Server

Source: https://pythe.us/quickstart

Command to start the Flask application.

```bash
python quickstart.py
```

--------------------------------

### Observe a metric with labels using keyword arguments

Source: https://pythe.us/labels

Alternatively, labels can be passed as keyword arguments to the `.labels()` method for a more concise syntax. This example observes a page hit with the 'GET' method.

```python
page_hit_total.labels(method='GET'}).inc()
```

--------------------------------

### Install Pytheus Rust Backend

Source: https://pythe.us/rust_backend

Install the Rust backend using pip. This package provides a faster alternative to the default Python backend.

```bash
pip install pytheus-backend-rs

```

--------------------------------

### Example Metrics Output

Source: https://pythe.us/fastapi

This is an example of the metrics output you will see after visiting the `/metrics` endpoint twice. The first visit collects data, and the second visit displays it.

```text
# HELP http_request_duration_seconds duration of the http request
# TYPE http_request_duration_seconds histogram
http_request_duration_seconds_bucket{method="GET",route="/metrics",status_code="200",le="0.005"} 1.0
http_request_duration_seconds_bucket{method="GET",route="/metrics",status_code="200",le="0.01"} 1.0
http_request_duration_seconds_bucket{method="GET",route="/metrics",status_code="200",le="0.025"} 1.0
http_request_duration_seconds_bucket{method="GET",route="/metrics",status_code="200",le="0.05"} 1.0
http_request_duration_seconds_bucket{method="GET",route="/metrics",status_code="200",le="0.1"} 1.0
http_request_duration_seconds_bucket{method="GET",route="/metrics",status_code="200",le="0.25"} 1.0
http_request_duration_seconds_bucket{method="GET",route="/metrics",status_code="200",le="0.5"} 1.0
http_request_duration_seconds_bucket{method="GET",route="/metrics",status_code="200",le="1"} 1.0
http_request_duration_seconds_bucket{method="GET",route="/metrics",status_code="200",le="2.5"} 1.0
http_request_duration_seconds_bucket{method="GET",route="/metrics",status_code="200",le="5"} 1.0
http_request_duration_seconds_bucket{method="GET",route="/metrics",status_code="200",le="10"} 1.0
http_request_duration_seconds_bucket{method="GET",route="/metrics",status_code="200",le="+Inf"} 1.0
http_request_duration_seconds_sum{method="GET",route="/metrics",status_code="200"} 0.0014027919969521463
http_request_duration_seconds_count{method="GET",route="/metrics",status_code="200"} 1.0
# HELP http_request_size_bytes http request size
# TYPE http_request_size_bytes histogram
# HELP http_response_size_bytes http response size
# TYPE http_response_size_bytes histogram
http_response_size_bytes_bucket{method="GET",route="/metrics",status_code="200",le="10.0"} 0.0
http_response_size_bytes_bucket{method="GET",route="/metrics",status_code="200",le="100.0"} 0.0
http_response_size_bytes_bucket{method="GET",route="/metrics",status_code="200",le="1000.0"} 1.0
http_response_size_bytes_bucket{method="GET",route="/metrics",status_code="200",le="10000.0"} 1.0
http_response_size_bytes_bucket{method="GET",route="/metrics",status_code="200",le="100000.0"} 1.0
http_response_size_bytes_bucket{method="GET",route="/metrics",status_code="200",le="1000000.0"} 1.0
http_response_size_bytes_bucket{method="GET",route="/metrics",status_code="200",le="10000000.0"} 1.0
http_response_size_bytes_bucket{method="GET",route="/metrics",status_code="200",le="100000000.0"} 1.0
http_response_size_bytes_bucket{method="GET",route="/metrics",status_code="200",le="+Inf"} 1.0
http_response_size_bytes_sum{method="GET",route="/metrics",status_code="200"} 296.0
http_response_size_bytes_count{method="GET",route="/metrics",status_code="200"} 1.0
```

--------------------------------

### Observe a metric with labels using a dictionary

Source: https://pythe.us/labels

To observe a metric with labels, use the `.labels()` method, passing a dictionary where keys are label names and values are label values. This example observes a page hit with the 'GET' method.

```python
page_hit_total.labels({'method': 'GET'}).inc()
```

--------------------------------

### Implement Prometheus metrics in a Flask application

Source: https://pythe.us/

Example demonstrating how to track request duration using histograms with context managers and decorators.

```python
import time
from flask import Flask, Response
from pytheus.metrics import Histogram
from pytheus.exposition import generate_metrics, PROMETHEUS_CONTENT_TYPE

app = Flask(__name__)

http_request_duration_seconds = Histogram(
    'http_request_duration_seconds', 'documenting the metric..'
)

@app.route('/metrics')
def metrics():
    data = generate_metrics()
    return Response(data, headers={'Content-Type': PROMETHEUS_CONTENT_TYPE})

# track time with the context manager
@app.route('/')
def home():
    with http_request_duration_seconds.time():
        return 'hello world!'

# alternatively you can also track time with the decorator shortcut
@app.route('/slow')
@http_request_duration_seconds
def slow():
    time.sleep(3)
    return 'hello world! from slow!'

app.run(host='0.0.0.0', port=8080)
```

--------------------------------

### Prometheus Metrics Output

Source: https://pythe.us/quickstart

Example of the raw text format returned by the /metrics endpoint.

```text
# HELP http_request_duration_seconds documenting the metric..
# TYPE http_request_duration_seconds histogram
http_request_duration_seconds_bucket{le="0.005"} 0.0
http_request_duration_seconds_bucket{le="0.01"} 0.0
http_request_duration_seconds_bucket{le="0.025"} 0.0
http_request_duration_seconds_bucket{le="0.05"} 0.0
http_request_duration_seconds_bucket{le="0.1"} 0.0
http_request_duration_seconds_bucket{le="0.25"} 0.0
http_request_duration_seconds_bucket{le="0.5"} 0.0
http_request_duration_seconds_bucket{le="1"} 0.0
http_request_duration_seconds_bucket{le="2.5"} 0.0
http_request_duration_seconds_bucket{le="5"} 0.0
http_request_duration_seconds_bucket{le="10"} 0.0
http_request_duration_seconds_bucket{le="+Inf"} 0.0
http_request_duration_seconds_sum 0.0
http_request_duration_seconds_count 0.0
```

--------------------------------

### Observe a metric with a different label value

Source: https://pythe.us/labels

You can observe the same metric with different label values. This example increments the `page_hit_total` by 3 for the 'POST' method.

```python
page_hit_total.labels({'method': 'POST'}).inc(3)
```

--------------------------------

### Create a Counter with a 'method' label

Source: https://pythe.us/labels

Use the `required_labels` parameter to define labels when creating a metric. This example creates a Counter that tracks page hits, differentiating them by HTTP method.

```python
from pytheus.metrics import Counter

page_hit_total = Counter(
    'page_hit_total',
    'number of time this page got it',
    required_labels=['method'],
)
```

--------------------------------

### Formatted metric output with labels

Source: https://pythe.us/labels

Metrics with labels are represented in a format that includes the label key-value pairs. This shows the output for the `page_hit_total` metric with 'GET' and 'POST' methods.

```text
page_hit_total{method="GET"} 1.0
page_hit_total{method="POST"} 3.0
```

--------------------------------

### Observe a cached metric instance

Source: https://pythe.us/labels

Once a metric instance is cached with specific labels, you can observe it directly without needing to specify the labels again. This increments the cached 'GET' method counter.

```python
page_hit_total_with_get.inc()
```

--------------------------------

### Cache metric instance with a set label

Source: https://pythe.us/labels

To avoid repeatedly calling `.labels()`, you can assign the result to a new variable. This creates a reusable instance of the metric with the 'GET' method label pre-set.

```python
page_hit_total_with_get = page_hit_total.labels({'method': 'GET'})
```

--------------------------------

### Initialize Custom Backend

Source: https://pythe.us/backend

Optional class method hook for one-time initialization steps.

```python
@classmethod
def _initialize(cls, config: "BackendConfig") -> None:
    # initialization steps
```

--------------------------------

### Initialize Summary Metric

Source: https://pythe.us/summary

Instantiate a Summary metric with a name and description. This metric tracks observed values and their counts.

```python
from pytheus.metrics import Summary

summary = Summary(name="http_request_duration_seconds", description="My description")
```

--------------------------------

### Use Summary Metrics

Source: https://pythe.us/quickref

Shows how to observe values and time code execution using Summary metrics.

```python
from pytheus.metrics import Summary

summary = Summary(name="my_summary", description="My description")

# observe a value
summary.observe(0.4)

# you can also time a piece of code, will set the duration in seconds to value when exited
with summary.time():
    do_something()

# tracking time can also be done as a decorator
@summary
def do_something():
    ...
```

--------------------------------

### load_backend

Source: https://pythe.us/backend

Initializes the backend for metric storage. This should be called before creating metrics.

```APIDOC
## load_backend

### Description
Initializes the backend used by Pytheus for metric storage. This operation must be performed before creating metrics.

### Method
Function Call

### Parameters
- **backend_class** (type[Backend] | None) - Optional - The backend class to use.
- **backend_config** (BackendConfig | None) - Optional - Configuration object for the backend.

### Configuration via Environment Variables
- **PYTHEUS_BACKEND_CLASS** - The class to import (e.g., pytheus.backends.redis.MultiProcessFileBackend).
- **PYTHEUS_BACKEND_CONFIG** - Path to a JSON file containing the backend configuration.
```

--------------------------------

### Load a Custom Backend

Source: https://pythe.us/backend

Switch the global backend implementation using load_backend.

```python
from pytheus.backends import load_backend
from pytheus.backends.redis import MultiProcessRedisBackend

load_backend(
    backend_class=MultiProcessRedisBackend,
    backend_config={"host": "127.0.0.1", "port": 6379},
)
```

--------------------------------

### Use Histogram Metrics

Source: https://pythe.us/quickref

Demonstrates creating histograms with custom buckets and observing values or timing code execution.

```python
from pytheus.metrics import Histogram

histogram = Histogram(name="my_histogram", description="My description")
# by default it will have the following buckets: (.005, .01, .025, .05, .1, .25, .5, 1, 2.5, 5, 10)
# note: the +Inf bucket will be added automatically, this is float('inf') in python

# create a histogram specifying buckets
histogram = Histogram(name="my_histogram", description="My description", buckets=(0.2, 1, 3))

# observe a value
histogram.observe(0.4)

# you can also time a piece of code, will set the duration in seconds to value when exited
with histogram.time():
    do_something()

# tracking time can also be done as a decorator
@histogram
def do_something():
    ...
```

--------------------------------

### Instantiate CollectorRegistry with Prefix

Source: https://pythe.us/registry

Create a `CollectorRegistry` with a specified prefix to prepend to all collected metrics.

```python
my_registry = CollectorRegistry(prefix="service_a")
```

--------------------------------

### Importing Exposition Utilities

Source: https://pythe.us/quickstart

Import necessary classes and functions for exposing metrics to Prometheus.

```python
from flask import Flask, Response  # we will need the Response class
from pytheus.exposition import generate_metrics, PROMETHEUS_CONTENT_TYPE
```

--------------------------------

### Check Default Backend

Source: https://pythe.us/backend

Verify the currently active backend implementation.

```python
from pytheus.metrics import Counter

counter = Counter('cache_hit_total', 'description')
print(counter._metric_value_backend.__class__)
# <class 'pytheus.backends.base.SingleProcessBackend'>
```

--------------------------------

### Initialize a Histogram

Source: https://pythe.us/histogram

Create a new Histogram instance with a name and description.

```python
from pytheus.metrics import Histogram

histogram = Histogram(name="http_request_duration_seconds", description="My description")
```

--------------------------------

### Manage Partial Labels

Source: https://pythe.us/quickref

Demonstrates how to define metrics with required labels and partially apply them using cacheable objects.

```python
from pytheus.metrics import Counter

# without labels
my_metric = Counter('metric_name', 'desc')
my_metric.inc()  # example for counter

# with labels
my_metric = Counter('metric_name', 'desc', required_labels=['req1', 'req2'])

my_metric.labels({'req1': '1', 'req2': '2'}).inc()  # you can pass all the labels at once
partial_my_metric = my_metric.labels({'req1': '1'})  # a cacheable object with one of the required labels already set
observable_my_metric = partial_my_metric.labels({'req2': '2'})  # finish setting the remaining values before observing
observable_my_metric.inc()

# alternatively to using a `dict` you can use `kwargs` (keyword arguments)
my_metric.labels(req1='1', req2='2')
```

--------------------------------

### Load RedisBackend

Source: https://pythe.us/rust_backend

Load the RedisBackend by specifying the backend class and configuration. Ensure the Redis server is running at the specified host and port.

```python
from pytheus_backend_rs import RedisBackend

load_backend(
    backend_class=RedisBackend,
    backend_config={"host": "127.0.0.1", "port": 6379},
)

```

--------------------------------

### Default Backend: SingleProcessBackend

Source: https://pythe.us/backend

Demonstrates the default thread-safe in-memory backend implementation.

```APIDOC
## Default Backend

The default backend used is a `SingleProcessBackend`. A thread-safe in-memory implementation that makes use of `threading.Lock`.

### Example Usage
```python
from pytheus.metrics import Counter

counter = Counter('cache_hit_total', 'description')
print(counter._metric_value_backend.__class__)
# <class 'pytheus.backends.base.SingleProcessBackend'>
```
```

--------------------------------

### Create a Counter Metric

Source: https://pythe.us/quickstart

Import and instantiate a Counter metric to track occurrences.

```python
from prometheus.metrics import Counter

page_hit_total = Counter('page_hit_total', 'documentation string')
```

--------------------------------

### Instantiating a Histogram Metric

Source: https://pythe.us/quickstart

Create a new Histogram instance with a name and documentation string.

```python
http_request_duration_seconds = Histogram(
    'http_request_duration_seconds', 'documenting the metric..'
)
```

--------------------------------

### Verify Backend Switch

Source: https://pythe.us/backend

Confirm that a newly created metric uses the configured backend.

```python
counter = Counter('cache_hit_total', 'description')
print(counter._metric_value_backend.__class__)
```

--------------------------------

### Instantiate a Counter Metric

Source: https://pythe.us/counter

Import and instantiate a Counter metric. This is useful for counting events like cache hits or connections.

```python
from pytheus.metrics import Counter

cache_hit_total = Counter(name='cache_hit_total', description='number of times the cache got it')
```

--------------------------------

### Instantiate CollectorRegistry

Source: https://pythe.us/registry

Create a new instance of `CollectorRegistry` to manage metrics.

```python
from pytheus.registry import CollectorRegistry

my_registry = CollectorRegistry()
```

--------------------------------

### Configure Default Labels

Source: https://pythe.us/quickref

Shows how to set default values for labels, which can be overridden during observation.

```python
from pytheus.metrics import Counter

# with default labels
my_metric = Counter('metric_name', 'desc', required_labels=['req1', 'req2'], default_labels={'req2': 'me_set!'})

my_metric.labels({'req1': '1'}).inc()  # as we have req2 as a default label we only need to set the remaining labels for observing
my_metric.labels({'req1': '1', 'req2': '2'})  # you can still override default labels!
```

--------------------------------

### Tracking Time with Context Manager

Source: https://pythe.us/quickstart

Use the .time() method as a context manager to measure execution time within a route.

```python
@app.route('/')
def home():
    with http_request_duration_seconds.time():
        return 'hello world!'
```

--------------------------------

### Use Counter Metrics

Source: https://pythe.us/quickref

Illustrates incrementing counters and using them to track exceptions via context managers or decorators.

```python
from pytheus.metrics import Counter

counter = Counter(name="my_counter", description="My description")

# increase by 1
counter.inc()

# increase by x
counter.inc(7)

# it is possible to count exceptions
with counter.count_exceptions():
    raise ValueError  # increases counter by 1

# you can specify which exceptions to watch for
with counter.count_exceptions((IndexError, ValueError)):
    raise ValueError. # increases counter by 1

# it is possible to use the counter as a decorator as a shortcut to count exceptions
@counter
def test():
    raise ValueError  # increases counter by 1 when called

# specifying which exceptions to look for also works with the decorator
@counter(exceptions=(IndexError, ValueError))
def test():
    raise ValueError  # increases counter by 1 when called
```

--------------------------------

### Track Time with Summary Context Manager

Source: https://pythe.us/summary

Utilize the time() context manager to automatically track the duration of a block of code in seconds. This is a convenient way to measure operation times.

```python
with summary.time():
    do_something()
```

--------------------------------

### Access the Global Registry

Source: https://pythe.us/quickstart

Import the global registry proxy to manage metrics.

```python
from pytheus.registry import REGISTRY
```

--------------------------------

### Load Backend Function Definition

Source: https://pythe.us/backend

This is the function definition for loading a backend. It can be configured via environment variables `PYTHEUS_BACKEND_CLASS` and `PYTHEUS_BACKEND_CONFIG`.

```python
def load_backend(
    backend_class: type[Backend] | None = None,
    backend_config: BackendConfig | None = None,
) -> None:


```

--------------------------------

### Labels method with kwargs support

Source: https://pythe.us/changelog

Demonstrates using the labels() method with keyword arguments for Prometheus metrics. This approach is an alternative to passing a dictionary.

```python
metric = Counter("counter", "desc", required_labels=["method"])

# dict approach
metric.labels({"method": "POST"})

# kwargs approach
metric.labels(method="POST")
```

--------------------------------

### Configure Python Logging

Source: https://pythe.us/rust_backend

Configure the Python logging library to view logs from the Rust backend. Ensure logging is set up before loading the backend.

```python
import logging
logging.basicConfig(level=logging.INFO)

```

--------------------------------

### Initialize a Gauge Metric

Source: https://pythe.us/gauge

Initialize a Gauge metric with a name and help string. This metric is suitable for measuring values that can fluctuate, such as memory usage or temperature.

```python
from pytheus.metrics import Gauge

gauge = Gauge('room_temperature', 'temperature in the living room')
```

--------------------------------

### Use Gauge Metrics

Source: https://pythe.us/quickref

Covers incrementing, decrementing, setting values, and tracking progress or duration with Gauges.

```python
from pytheus.metrics import Gauge

gauge = Gauge(name="my_gauge", description="My description")

# increase by 1
gauge.inc()

# increase by x
gauge.inc(7)

# decrease by 1
gauge.dec()

# set a specific value
gauge.set(7)

# set to current unix timestamp
gauge.set_to_current_time()

# it is possible to track progress so that when entered increases the value by 1, and when exited decreases it
with gauge.track_inprogress():
    do_something()

# you can also time a piece of code, will set the duration in seconds to value when exited
with gauge.time():
    do_something()

# tracking time can also be done as a decorator
@gauge
def do_something():
    ...

# tracking progress is also available via decorator with a flag
@gauge(track_inprogress=True)
def do_something():
    ...
```

--------------------------------

### Observe Metrics with Partial Labels

Source: https://pythe.us/labels

Complete the label set for a partial metric instance to record observations.

```python
cache_hit_count_total_system_a.labels({'origin': 'function_a'}).inc()
cache_hit_count_total_system_a.labels({'origin': 'function_b'}).inc()

cache_hit_count_total_system_b.labels({'origin': 'function_a'}).inc()
```

--------------------------------

### Define Metrics with Required Labels

Source: https://pythe.us/labels

Initialize a Counter with required labels and create partial metric instances by pre-setting a subset of those labels.

```python
cache_hit_count_total = Counter(
    'cache_hit_count_total',
    'number of time the cache got it',
    required_labels=['system', 'origin'],
)

cache_hit_count_total_system_a = cache_hit_count_total.labels({'system': 'a'})
cache_hit_count_total_system_b = cache_hit_count_total.labels({'system': 'b'})
```

--------------------------------

### Creating Metrics Endpoint

Source: https://pythe.us/quickstart

Define a route that returns the generated metrics data with the correct content type header.

```python
@app.route('/metrics')
def metrics():
    data = generate_metrics()
    return Response(data, headers={'Content-Type': PROMETHEUS_CONTENT_TYPE})
```

--------------------------------

### Define a Custom Backend Protocol

Source: https://pythe.us/backend

The structure required for implementing a custom backend class.

```python
class Backend(Protocol):
    def __init__(
        self,
        config: BackendConfig,
        metric: "_Metric",
        ,
    ) -> None:
        ...

    def inc(self, value: float) -> None:
        ...

    def dec(self, value: float) -> None:
        ...

    def set(self, value: float) -> None:
        ...

    def get(self) -> float:
        ...
```

--------------------------------

### Configure Redis Backend Expiration

Source: https://pythe.us/backend

Set the time-to-live for keys in Redis using the load_backend function.

```python
# 5 min
load_backend(MultiProcessRedisBackend, {"expire_key_time": 300})
```

--------------------------------

### Create Flask API Service

Source: https://pythe.us/quickstart

A basic Flask application with standard and delayed endpoints.

```python
import time
from flask import Flask

app = Flask(__name__)

# normal endpoint
@app.route('/')
def home():
    return 'hello world!'

# slowed endpoint with the `time` library
@app.route('/slow')
def slow():
    time.sleep(3)
    return 'hello world! from slow!'

app.run(host='0.0.0.0', port=8080)
```

--------------------------------

### Observe a Value with Summary

Source: https://pythe.us/summary

Use the observe() method to record a single observed value. This updates the total sum and count for the Summary metric.

```python
summary.observe(0.4)
```

--------------------------------

### Configure Histogram Buckets

Source: https://pythe.us/histogram

Define custom buckets for the histogram. The +Inf bucket is added automatically.

```python
(.005, .01, .025, .05, .1, .25, .5, 1, 2.5, 5, 10)
```

```python
customized_buckets = (0.1, 0.3, 2, 5)
histogram = Histogram(
    name="http_request_duration_seconds",
    description="My description",
    buckets=customized_buckets,
)
```

--------------------------------

### Expose Metrics Endpoint in FastAPI

Source: https://pythe.us/fastapi

Create a `/metrics` endpoint in your FastAPI application to expose the collected metrics. This endpoint uses `generate_metrics` from `pytheus.exposition` and returns a `PlainTextResponse`.

```python
from fastapi.responses import PlainTextResponse
from pytheus.exposition import generate_metrics


@app.get('/metrics', response_class=PlainTextResponse)
async def pytheus_metrics():
    return generate_metrics()
```

--------------------------------

### Backend Protocol Definition

Source: https://pythe.us/backend

Defines the interface for backend implementations, including initialization and value manipulation methods.

```APIDOC
## Backend Protocol

The backend protocol is as follows:
```python
class Backend(Protocol):
    def __init__(
        self,
        config: BackendConfig,
        metric: "_Metric",
        histogram_bucket: str | None = None,
    ) -> None:
        ...

    def inc(self, value: float) -> None:
        ...

    def dec(self, value: float) -> None:
        ...

    def set(self, value: float) -> None:
        ...

    def get(self) -> float:
        ...
```

When creating a custom backend the `__init__` method has to accept:
  * `config: BackendConfig`: possible configuration values for initializing it
  * `metric: _Metric`: the metric that is creating the instance of this backend for a value
  * `histogram_bucket: str | None`: bucket name in case it's instanced for an histogram

You don't have to use these values in your implementation, they just need to be accepted.
Note
`BackendConfig` is of type `dict[str, Any]`
`_Metric` is any of `Counter`, `Gauge`, `Histogram`

### Initialization Hook

To achieve custom initialization steps on import, use the `_initialize` class method hook:
```python
@classmethod
def _initialize(cls, config: "BackendConfig") -> None:
    # initialization steps
```

### `_generate_samples` Class Method

A `Backend` can have a `_generate_samples` class method for generating metrics more efficiently, for example, by using pipelines in `MultiProcessRedisBackend`.
```

--------------------------------

### Attempt to observe a metric without required labels

Source: https://pythe.us/labels

Observing a metric that has required labels without providing those labels will result in an `UnobservableMetricException`. Ensure all required labels are supplied.

```python
page_hit_total.inc()
```

--------------------------------

### Importing Histogram Metric

Source: https://pythe.us/quickstart

Import the Histogram class from the pytheus.metrics module.

```python
from pytheus.metrics import Histogram
```

--------------------------------

### Register Metric with Custom Registry

Source: https://pythe.us/registry

Create a metric and associate it with a specific `CollectorRegistry` instance.

```python
my_registry = CollectorRegistry()
counter = Counter('cache_hit_total', 'description', registry=my_registry)
```

--------------------------------

### Generate Metrics with Custom Registry

Source: https://pythe.us/registry

Use the `generate_metrics` function with a specific `CollectorRegistry` instance to collect and format metrics.

```python
from pytheus.exposition import generate_metrics

my_registry = CollectorRegistry()
generate_metrics(my_registry)
```

--------------------------------

### Tracking Time with Decorator

Source: https://pythe.us/quickstart

Use the metric instance as a decorator to automatically track the execution time of a function.

```python
@app.route('/slow')
@http_request_duration_seconds
def slow():
    time.sleep(3)
    return 'hello world! from slow!'
```

--------------------------------

### Generate Metrics for Exposition

Source: https://pythe.us/quickstart

Use generate_metrics to retrieve current metrics in the Prometheus text format.

```python
from pytheus.exposition import generate_metrics

metrics_data = generate_metrics()
```

--------------------------------

### Create Counter Without Global Registry

Source: https://pythe.us/registry

Create a new metric that is not automatically added to the global REGISTRY by passing `registry=None`.

```python
counter = Counter('cache_hit_total', 'description', registry=None)
```

--------------------------------

### Time Code Execution with Gauge

Source: https://pythe.us/gauge

Use the `time` context manager to measure the duration of a code block in seconds. The Gauge value will be set to the execution time.

```python
with gauge.time():
    do_something()
```

--------------------------------

### Use Summary as a Decorator

Source: https://pythe.us/summary

Apply the Summary metric as a decorator to a function to automatically time its execution. This provides syntactic sugar over the time() context manager.

```python
@summary
def do_something():
    ...
```

--------------------------------

### Track In-Progress Operations with Gauge

Source: https://pythe.us/gauge

Use the `track_inprogress` context manager to automatically increment the Gauge on entry and decrement on exit. This is ideal for monitoring the number of concurrent operations.

```python
with gauge.track_inprogress():
    do_something()
```

--------------------------------

### Track Execution Time

Source: https://pythe.us/histogram

Measure the duration of a block of code using a context manager or decorator.

```python
with histogram.time():
    do_something()
```

```python
@histogram
def do_something():
    ...
```

--------------------------------

### Registry Protocol Definition

Source: https://pythe.us/registry

Defines the required methods for a custom registry: `register`, `unregister`, and `collect`.

```python
class Registry(Protocol):
    prefix: str | None

    def register(self, collector: Collector) -> None:
        ...

    def unregister(self, collector: Collector) -> None:
        ...

    def collect(self) -> Iterable:
        ...
```

--------------------------------

### Create a Histogram with default labels

Source: https://pythe.us/labels

Define default values for required labels using the `default_labels` parameter. This `Histogram` metric for HTTP request duration will default the 'service' label to 'main_service'.

```python
http_request_duration_seconds = Histogram(
    'http_request_duration_seconds',
    'documenting the metric..',
    required_labels=['service'],
    default_labels={'service': 'main_service'},
)
```

--------------------------------

### Observe a metric with default labels

Source: https://pythe.us/labels

Metrics configured with `default_labels` can be observed directly for the required labels, as the default value will be used. This observes the histogram with the default 'service' label.

```python
http_request_duration_seconds.observe(1)
```

--------------------------------

### Multiprocess Backend: MultiProcessRedisBackend

Source: https://pythe.us/backend

Details the Redis-based backend for multi-process environments, emphasizing atomic operations and configuration options.

```APIDOC
## Multiprocess Backend

The library includes a redis based implementation of the backend for supporting multi process python services: `MultiProcessRedisBackend`.
This makes use of the `INCRBYFLOAT` & `HINCRBYFLOAT` redis operations that are `ATOMIC` and as redis is single-threaded it means that even if multiple clients try to update the same value we will end up with the correct one.

### Configuration

*   **`key_prefix`**: (Optional) A prefix to be added to all stored metrics, useful when sharing a Redis server across multiple services.
*   **`expire_key_time`**: (Optional) The time in seconds keys will stay alive in Redis. Defaults to 1 hour and is refreshed on each operation.

### Example Configuration

```python
# Set expire_key_time to 5 minutes
from pytheus.backends import load_backend
from pytheus.backends.redis import MultiProcessRedisBackend

load_backend(MultiProcessRedisBackend, {"expire_key_time": 300})
```

### Loading a Different Backend

To load a different backend, use the `load_backend` function.

### Example: Loading `MultiProcessRedisBackend`

```python
from pytheus.backends import load_backend
from pytheus.backends.redis import MultiProcessRedisBackend

load_backend(
    backend_class=MultiProcessRedisBackend,
    backend_config={"host": "127.0.0.1", "port": 6379},
)

# Metrics will now use the configured backend
from pytheus.metrics import Counter
counter = Counter('cache_hit_total', 'description')
print(counter._metric_value_backend.__class__)
```
```

--------------------------------

### Manually Register Metric

Source: https://pythe.us/registry

Register an already created metric with a `CollectorRegistry` instance using the `register` method.

```python
my_registry.register(counter)
```

--------------------------------

### Implement a Custom Collector

Source: https://pythe.us/custom_collector

Inherit from CustomCollector and define the collect method. Ensure automatic registration is disabled by setting registry=None.

```python
from pytheus.metrics import Counter, CustomCollector
from pytheus.registry import REGISTRY

class MyCustomCollector(CustomCollector):
    def collect():
        # note that we are disabling automatic registering
        counter = Counter('name', 'desc', registry=None)
        counter.inc()
        yield counter

REGISTRY.register(MyCustomCollector())
```

--------------------------------

### Add Pytheus Middleware to FastAPI

Source: https://pythe.us/fastapi

Import and add the PytheusMiddlewareASGI to your FastAPI application. This middleware automatically collects HTTP request metrics.

```python
from fastapi import FastAPI
from pytheus.middleware import PytheusMiddlewareASGI


app = FastAPI()
app.add_middleware(PytheusMiddlewareASGI)
```

--------------------------------

### Observe Histogram Values

Source: https://pythe.us/histogram

Record a single observation using the observe method.

```python
histogram.observe(0.4)
```

--------------------------------

### Register a Custom Collector

Source: https://pythe.us/quickstart

Manually register a collector object to the global registry.

```python
REGISTRY.register(mycollector)
```

--------------------------------

### Generate Metrics in FastAPI Endpoint

Source: https://pythe.us/rust_backend

Define a FastAPI endpoint to generate metrics. For asynchronous applications, it's recommended to call `generate_metrics` within a ThreadPool. FastAPI handles this automatically for 'def' defined endpoints.

```python
@app.get('/metrics')
def pytheus_metrics():
    return generate_metrics()

```

--------------------------------

### Cache Metric Instances with Partial Labels

Source: https://pythe.us/labels

Store partially labeled metrics in variables to avoid repetitive label definitions.

```python
cache_hit_count_total_system_a_with_origin.labels({'origin': 'function_a'})

cache_hit_count_total_system_a_with_origin.inc()
```

--------------------------------

### Set Default Global Registry

Source: https://pythe.us/registry

Replace the default global registry with a custom `CollectorRegistry` instance using `REGISTRY.set_registry()`.

```python
from pytheus.registry import REGISTRY, CollectorRegistry

my_registry = CollectorRegistry(prefix='service_a')
REGISTRY.set_registry(my_registry)
```

--------------------------------

### Count Exceptions within a `with` Statement

Source: https://pythe.us/counter

Use the `count_exceptions()` context manager to automatically increment a Counter when an exception occurs within the `with` block. Assumes a Counter named `counter` is defined.

```python
with counter.count_exceptions():
    raise ValueError  # increases counter by 1
```

--------------------------------

### Set Prometheus Content-Type Header

Source: https://pythe.us/quickstart

Use the constant provided by the library to set the correct Content-Type header for metrics endpoints.

```python
from pytheus.exposition import PROMETHEUS_CONTENT_TYPE

headers = {'Content-Type': PROMETHEUS_CONTENT_TYPE}
```

--------------------------------

### Gauge as Decorator for In-Progress Tracking

Source: https://pythe.us/gauge

Use the Gauge decorator with `track_inprogress=True` to automatically manage the Gauge value for in-progress operations. This is syntactic sugar for the `track_inprogress` context manager.

```python
@gauge(track_inprogress=True)
def do_something():
    ...
```

--------------------------------

### Gauge as Decorator for Timing

Source: https://pythe.us/gauge

Apply the Gauge as a decorator to automatically time the execution of a function. This provides syntactic sugar for the `time()` context manager.

```python
@gauge
def do_something():
    ...
```

--------------------------------

### Override default labels

Source: https://pythe.us/labels

You can override the `default_labels` by explicitly setting a different value when calling the `.labels()` method. This creates a metric instance for a 'side_service'.

```python
http_request_duration_seconds_side_service = http_request_duration_seconds.labels(
    {'service': 'side_service'}
)
```

--------------------------------

### Count Exceptions as a Decorator

Source: https://pythe.us/counter

Use the Counter as a decorator to automatically count exceptions raised by a function. This is syntactic sugar for `count_exceptions()`.

```python
@counter
def my_func():
    raise ValueError  # increases counter by 1 when it raises
```

--------------------------------

### Increment Counter Value by 1

Source: https://pythe.us/counter

Increment the value of a Counter metric by 1 using the `inc()` method. This is the default increment amount.

```python
# increases value by 1
cache_hit_total.inc()
```

--------------------------------

### Count Specific Exceptions as a Decorator

Source: https://pythe.us/counter

Specify a tuple of Exception types to be counted when using the Counter as a decorator. Exceptions not in the list will not increment the counter.

```python
@counter(exceptions=(IndexError, ValueError))
def my_func():
    raise KeyError  # won't increase when raised as it's not in the list
```

--------------------------------

### Count Specific Exceptions within a `with` Statement

Source: https://pythe.us/counter

Specify a tuple of Exception types to be counted by `count_exceptions()`. Exceptions not in the tuple will not increment the counter.

```python
with counter.count_exceptions((IndexError, ValueError)):
    raise KeyError  # does not increase as it's not included in the list
```

--------------------------------

### Increment a Counter Metric

Source: https://pythe.us/quickstart

Use the inc() method to increment the counter value, optionally by a specific amount.

```python
page_hit_total.inc()
# or specify the amount
page_hit_total.inc(2)
```

--------------------------------

### Increment Counter Value by a Specific Amount

Source: https://pythe.us/counter

Increment the value of a Counter metric by a specified positive amount using `inc(amount)`. Counters cannot decrease in value.

```python
# increases value by 7
cache_hit_total.inc(7)
```

--------------------------------

### Set Gauge Value

Source: https://pythe.us/gauge

Set the Gauge value directly to a specific number or the current Unix timestamp. Useful for absolute value assignments.

```python
gauge.set(3)
```

```python
gauge.set_to_current_time()
```

--------------------------------

### Increment Gauge Value

Source: https://pythe.us/gauge

Increment the Gauge value by 1 or a specified amount. Use this to increase the metric's value when a specific event occurs.

```python
# increase by 1
gauge.inc()
```

```python
# increase by 7
gauge.inc(7)
```

--------------------------------

### Decrement Gauge Value

Source: https://pythe.us/gauge

Decrement the Gauge value by 1 or a specified amount. Use this to decrease the metric's value.

```python
# decrease by 1
gauge.dec()
```

```python
# decrease by 7
gauge.dec(7)
```
Tokens:
Raw