### Start Flask Server Source: https://github.com/llandy3d/pytheus/blob/main/docs/quickstart.md Command to execute the quickstart script. ```bash python quickstart.py ``` -------------------------------- ### Setup Project Environment Source: https://github.com/llandy3d/pytheus/blob/main/docs/quickstart.md Commands to create a directory, 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 Source: https://context7.com/llandy3d/pytheus/llms.txt Commands to install the core library and optional backends. ```bash # Basic installation pip install pytheus # With Redis backend support pip install pytheus[redis] # With Rust backend support pip install pytheus-backend-rs ``` -------------------------------- ### Run Application Source: https://context7.com/llandy3d/pytheus/llms.txt Start the application server. ```python if __name__ == '__main__': app.run(host='0.0.0.0', port=8080) ``` -------------------------------- ### Install Pytheus and Backends Source: https://github.com/llandy3d/pytheus/blob/main/README.md Commands to install the core library and optional backends for multiprocessing support. ```bash pip install pytheus ``` ```python pip install redis # or pip install pytheus[redis] ``` ```bash pip install pytheus-backend-rs ``` -------------------------------- ### Implement Flask Application Metrics Source: https://context7.com/llandy3d/pytheus/llms.txt Example setup for exposing metrics in a Flask application, including commented-out configuration for Redis backends. ```python import time from flask import Flask, Response from pytheus.metrics import Counter, Gauge, Histogram from pytheus.exposition import generate_metrics, PROMETHEUS_CONTENT_TYPE # Optional: Enable Redis backend for multi-process # from pytheus.backends import load_backend # from pytheus.backends.redis import MultiProcessRedisBackend # load_backend(MultiProcessRedisBackend, {'host': 'localhost', 'port': 6379}) app = Flask(__name__) ``` -------------------------------- ### Backend Log Output Source: https://github.com/llandy3d/pytheus/blob/main/docs/rust_backend.md Example output showing thread initialization and backend status. ```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 ``` -------------------------------- ### Initialize custom backend with _initialize Source: https://github.com/llandy3d/pytheus/blob/main/docs/backend.md Use the _initialize class method for one-time setup steps required by the backend. ```python @classmethod def _initialize(cls, config: "BackendConfig") -> None: # initialization steps ``` -------------------------------- ### Install Pytheus Rust Backend Source: https://github.com/llandy3d/pytheus/blob/main/docs/index.md Install the Pytheus Rust backend for enhanced multiprocessing support. This requires a separate package. ```bash pip install pytheus-backend-rs ``` -------------------------------- ### Install Pytheus Source: https://github.com/llandy3d/pytheus/blob/main/docs/index.md Install the Pytheus library using pip. For Redis backend support, install with the 'redis' extra. ```bash pip install pytheus ``` ```bash pip install redis # or everything in one command pip install pytheus[redis] ``` -------------------------------- ### Example Metrics Output Source: https://github.com/llandy3d/pytheus/blob/main/docs/fastapi.md Sample output format for metrics collected by the middleware, including duration and size histograms. ```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 ``` -------------------------------- ### Initialize Histogram Metric Source: https://github.com/llandy3d/pytheus/blob/main/docs/histogram.md Instantiate a Histogram metric with a name and description. This is the basic setup for tracking observations. ```python from pytheus.metrics import Histogram histogram = Histogram(name="http_request_duration_seconds", description="My description") ``` -------------------------------- ### Expose Metrics via Flask Source: https://context7.com/llandy3d/pytheus/llms.txt Provides an example of creating a /metrics endpoint in a Flask application to serve Prometheus-compatible metrics. ```python from flask import Flask, Response from pytheus.metrics import Counter, Histogram from pytheus.exposition import generate_metrics, PROMETHEUS_CONTENT_TYPE app = Flask(__name__) # Define metrics requests_total = Counter('requests_total', 'Total requests') request_duration = Histogram('request_duration_seconds', 'Request duration') @app.route('/metrics') def metrics(): """Endpoint for Prometheus to scrape.""" data = generate_metrics() return Response(data, headers={'Content-Type': PROMETHEUS_CONTENT_TYPE}) @app.route('/') @request_duration def home(): requests_total.inc() return 'Hello World!' ``` -------------------------------- ### Prometheus metrics output format Source: https://github.com/llandy3d/pytheus/blob/main/docs/quickstart.md Example of the raw metrics data format exposed at 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 ``` -------------------------------- ### Basic Flask App with Pytheus Metrics Source: https://github.com/llandy3d/pytheus/blob/main/docs/index.md This example demonstrates a Flask application instrumented with Pytheus to expose HTTP request duration metrics. It uses a Histogram metric and exposes them via a /metrics endpoint. Metrics can be tracked using a context manager or a decorator. ```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) ``` -------------------------------- ### View created time series Source: https://github.com/llandy3d/pytheus/blob/main/docs/labels.md Example output of time series data generated by the metrics. ```text page_hit_total{method="GET"} 1.0 page_hit_total{method="POST"} 3.0 ``` -------------------------------- ### Initialize a Summary metric Source: https://github.com/llandy3d/pytheus/blob/main/docs/summary.md Import the Summary class and instantiate it with a name and description. ```python from pytheus.metrics import Summary summary = Summary(name="http_request_duration_seconds", description="My description") ``` -------------------------------- ### Configure Metric Backends in Python Source: https://context7.com/llandy3d/pytheus/llms.txt Switch between default in-memory storage and Redis-backed storage for multi-process support using load_backend. ```python from pytheus.backends import load_backend from pytheus.backends.redis import MultiProcessRedisBackend from pytheus.metrics import Counter # Default: SingleProcessBackend (thread-safe, in-memory) counter = Counter('my_counter', 'description') print(counter._metric_value_backend.__class__) # # Load Redis backend for multi-process support load_backend( backend_class=MultiProcessRedisBackend, backend_config={ 'host': '127.0.0.1', 'port': 6379, 'expire_key_time': 3600 # Optional: key expiry in seconds (default: 1 hour) } ) # Now metrics use Redis backend redis_counter = Counter('redis_counter', 'Redis-backed counter') print(redis_counter._metric_value_backend.__class__) # # Alternative: Configure via environment variables # export PYTHEUS_BACKEND_CLASS="pytheus.backends.redis.MultiProcessRedisBackend" # export PYTHEUS_BACKEND_CONFIG="./config.json" # # config.json: # {"host": "127.0.0.1", "port": 6379} ``` -------------------------------- ### Import exposition utilities Source: https://github.com/llandy3d/pytheus/blob/main/docs/quickstart.md Import the generate_metrics function and PROMETHEUS_CONTENT_TYPE helper for exposing metrics. ```python from flask import Flask, Response # we will need the Response class from pytheus.exposition import generate_metrics, PROMETHEUS_CONTENT_TYPE ``` -------------------------------- ### Implement Summary Metrics Source: https://github.com/llandy3d/pytheus/blob/main/docs/quickref.md Demonstrates observing values and timing 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(): ... ``` -------------------------------- ### Create a Default Pytheus Flask Server Source: https://github.com/llandy3d/pytheus/blob/main/README.md Demonstrates setting up a basic Flask application with a Histogram metric and exposing it via a /metrics endpoint. ```python import time from flask import Flask from pytheus.metrics import Histogram from pytheus.exposition import generate_metrics app = Flask(__name__) histogram = Histogram('page_visits_latency_seconds', 'used for testing') # this is the endpoint that prometheus will use to scrape the metrics @app.route('/metrics') def metrics(): return generate_metrics() # track time with the context manager @app.route('/') def home(): with histogram.time(): return 'hello world!' # you can also track time with the decorator shortcut @app.route('/slow') @histogram def slow(): time.sleep(3) return 'hello world! from slow!' app.run(host='0.0.0.0', port=8080) ``` -------------------------------- ### Configure Default and Partial Labels Source: https://context7.com/llandy3d/pytheus/llms.txt Shows how to define default labels to reduce repetition and use partial labels for incremental metric building. ```python # Use default labels to avoid repetition http_duration = Histogram( 'http_request_duration_seconds', 'request duration', required_labels=['service', 'method'], default_labels={'service': 'main_api'} ) # Only need to set non-default labels http_duration.labels(method='GET').observe(0.5) # Override default labels when needed http_duration.labels(service='auth_service', method='POST').observe(0.8) # Partial labels for incremental building cache_hits = Counter( 'cache_hit_total', 'cache hits', required_labels=['system', 'operation'] ) # Create partial metric with one label set system_a = cache_hits.labels(system='a') system_b = cache_hits.labels(system='b') # Complete labels when observing system_a.labels(operation='read').inc() system_a.labels(operation='write').inc() system_b.labels(operation='read').inc() ``` -------------------------------- ### Configure Logging Source: https://github.com/llandy3d/pytheus/blob/main/docs/rust_backend.md Initialize the standard Python logging library to capture backend initialization and error logs. ```python import logging logging.basicConfig(level=logging.INFO) ``` -------------------------------- ### Configure Backend Directly in Python Source: https://github.com/llandy3d/pytheus/blob/main/README.md Load a backend and its configuration directly within Python code, which takes precedence over environment variables. ```python from pytheus.metrics import Counter 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 } ) # Notice that if you simply call load_backend(), it would reload config from the environment. # load_backend() is called automatically at package import, that's why we didn't need to call it # directly in the previous example counter = Counter( name="my_metric", description="My description", required_labels=["label_a", "label_b"], ) print(counter._metric_value_backend.__class__) # print(counter._metric_value_backend.config) # {'host': '127.0.0.1', 'port': 6379} ``` -------------------------------- ### Implement Gauge Metrics Source: https://github.com/llandy3d/pytheus/blob/main/docs/quickref.md Demonstrates setting values, tracking progress, and timing code execution 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(): ... ``` -------------------------------- ### Set Metric Labels Source: https://context7.com/llandy3d/pytheus/llms.txt Demonstrates setting labels using dictionaries, keyword arguments, and caching labeled metric instances for reuse. ```python # Set labels using dict page_hit_total.labels({'method': 'GET', 'status': '200'}).inc() # Set labels using kwargs page_hit_total.labels(method='POST', status='201').inc(3) # Cache labeled metric instance for reuse get_200 = page_hit_total.labels(method='GET', status='200') get_200.inc() get_200.inc(5) ``` -------------------------------- ### Implement Histogram Metrics Source: https://github.com/llandy3d/pytheus/blob/main/docs/quickref.md Shows how to define custom buckets and observe values or time 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(): ... ``` -------------------------------- ### Manage Metric Registries Source: https://context7.com/llandy3d/pytheus/llms.txt Demonstrates creating custom registries, registering metrics, and setting a global registry. ```python from pytheus.registry import REGISTRY, CollectorRegistry from pytheus.metrics import Counter from pytheus.exposition import generate_metrics # Metrics automatically register to global REGISTRY counter1 = Counter('app_requests_total', 'Total requests') # Create metric without auto-registration counter2 = Counter('temp_counter', 'Temporary', registry=None) # Create custom registry with prefix service_registry = CollectorRegistry(prefix='myservice') # Register metric to custom registry api_counter = Counter('api_calls_total', 'API calls', registry=service_registry) # Generate metrics from custom registry # Output: myservice_api_calls_total instead of api_calls_total service_metrics = generate_metrics(service_registry) # Replace default global registry custom_registry = CollectorRegistry(prefix='app') REGISTRY.set_registry(custom_registry) # Manually register/unregister collectors REGISTRY.register(api_counter) REGISTRY.unregister(api_counter) ``` -------------------------------- ### Configure Default Labels Source: https://github.com/llandy3d/pytheus/blob/main/docs/quickref.md Shows how to set default values for labels, allowing for partial application 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! ``` -------------------------------- ### Initialize a Gauge Source: https://github.com/llandy3d/pytheus/blob/main/docs/gauge.md Create a new Gauge instance by providing a name and a description. ```python from pytheus.metrics import Gauge gauge = Gauge('room_temperature', 'temperature in the living room') ``` -------------------------------- ### Use Default SingleProcessBackend Source: https://github.com/llandy3d/pytheus/blob/main/README.md Demonstrates the default metric backend, SingleProcessBackend, when creating a Counter. This backend is used out-of-the-box. ```python from pytheus.metrics import Counter counter = Counter( name="my_metric", description="My description", required_labels=["label_a", "label_b"], ) print(counter._metric_value_backend.__class__) # print(counter._metric_value_backend.config) # {} ``` -------------------------------- ### Create a Counter Metric Source: https://github.com/llandy3d/pytheus/blob/main/docs/quickstart.md Import and instantiate a Counter metric to track occurrences. ```python from prometheus.metrics import Counter page_hit_total = Counter('page_hit_total', 'documentation string') ``` -------------------------------- ### load_backend Source: https://github.com/llandy3d/pytheus/blob/main/docs/backend.md Configures the global backend implementation used by metrics in the Pytheus library. ```APIDOC ## load_backend ### Description Sets the backend class and configuration to be used for all subsequently created metrics. ### Parameters #### Request Body - **backend_class** (class) - Required - The backend implementation class (e.g., MultiProcessRedisBackend). - **backend_config** (dict[str, Any]) - Required - Configuration dictionary for the backend (e.g., host, port, key_prefix, expire_key_time). ### Request Example ```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, "expire_key_time": 300}, ) ``` ``` -------------------------------- ### Manage Partial Labels in Pytheus Source: https://github.com/llandy3d/pytheus/blob/main/docs/quickref.md Demonstrates how to define required labels and partially apply them to create cacheable metric 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') ``` -------------------------------- ### Create CollectorRegistry Instance Source: https://github.com/llandy3d/pytheus/blob/main/docs/registry.md Instantiate a `CollectorRegistry` to manage metrics. Optionally, set a `prefix` to prepend to all collected metric names. ```python from pytheus.registry import CollectorRegistry my_registry = CollectorRegistry() ``` ```python my_registry = CollectorRegistry(prefix="service_a") ``` -------------------------------- ### Instantiate Histogram metric Source: https://github.com/llandy3d/pytheus/blob/main/docs/quickstart.md Create a Histogram instance with a name and a mandatory documentation string. ```python http_request_duration_seconds = Histogram( 'http_request_duration_seconds', 'documenting the metric..' ) ``` -------------------------------- ### Import Histogram metric class Source: https://github.com/llandy3d/pytheus/blob/main/docs/quickstart.md Import the Histogram class from the pytheus library to begin tracking duration metrics. ```python from pytheus.metrics import Histogram ``` -------------------------------- ### Observe metrics using labels Source: https://github.com/llandy3d/pytheus/blob/main/docs/labels.md Provide label values using a dictionary or keyword arguments to observe metrics. ```python page_hit_total.labels({'method': 'GET'}).inc() ``` ```python page_hit_total.labels(method='GET'}).inc() ``` ```python page_hit_total.labels({'method': 'POST'}).inc(3) ``` -------------------------------- ### Instantiate a Counter Metric Source: https://github.com/llandy3d/pytheus/blob/main/docs/counter.md Import and instantiate a Counter metric to track increasing values. 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') ``` -------------------------------- ### Load a custom backend Source: https://github.com/llandy3d/pytheus/blob/main/docs/backend.md Use load_backend to switch the global backend implementation, such as switching to Redis for multi-process support. ```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}, ) ``` ```python counter = Counter('cache_hit_total', 'description') print(counter._metric_value_backend.__class__) ``` -------------------------------- ### Define a custom Backend protocol Source: https://github.com/llandy3d/pytheus/blob/main/docs/backend.md Implement this protocol to create custom storage backends for metrics. ```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: ... ``` -------------------------------- ### Load Rust Backend Source: https://github.com/llandy3d/pytheus/blob/main/docs/rust_backend.md Configure the library to use the RedisBackend by passing the class and connection configuration to load_backend. ```python from pytheus_backend_rs import RedisBackend load_backend( backend_class=RedisBackend, backend_config={"host": "127.0.0.1", "port": 6379}, ) ``` -------------------------------- ### Instrument with Summary Source: https://context7.com/llandy3d/pytheus/llms.txt Use Summaries to track latencies or event sizes without predefined buckets. ```python from pytheus.metrics import Summary # Create a summary request_latency = Summary( name='request_latency_seconds', description='Request latency in seconds' ) # Observe a value request_latency.observe(0.4) # Time a piece of code with request_latency.time(): process_data() # Use as decorator @request_latency def api_call(): return fetch_api_response() ``` -------------------------------- ### Access the Global Registry Source: https://github.com/llandy3d/pytheus/blob/main/docs/quickstart.md Import the global registry proxy to manage metrics. ```python from pytheus.registry import REGISTRY ``` -------------------------------- ### Implement Counter Metrics Source: https://github.com/llandy3d/pytheus/blob/main/docs/quickref.md Covers basic incrementing and exception tracking using 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 ``` -------------------------------- ### Use Redis Backend with Environment Configuration Source: https://github.com/llandy3d/pytheus/blob/main/README.md Verify the MultiProcessRedisBackend is active after setting environment variables and loading the configuration. ```python from pytheus.metrics import Counter counter = Counter( name="my_metric", description="My description", required_labels=["label_a", "label_b"], ) print(counter._metric_value_backend.__class__) # print(counter._metric_value_backend.config) # {"host": "127.0.0.1", "port": 6379} ``` -------------------------------- ### Define Custom Backend Protocol Source: https://github.com/llandy3d/pytheus/blob/main/README.md Implement the Backend protocol by defining the required methods for custom backend creation. ```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: ... ``` -------------------------------- ### Configure Redis Backend via Environment Variables Source: https://github.com/llandy3d/pytheus/blob/main/README.md Set environment variables to configure Pytheus to use the MultiProcessRedisBackend and specify a configuration file. ```bash export PYTHEUS_BACKEND_CLASS="pytheus.backends.redis.MultiProcessRedisBackend" export PYTHEUS_BACKEND_CONFIG="./config.json" ``` -------------------------------- ### Integrate Rust-powered Backend Source: https://context7.com/llandy3d/pytheus/llms.txt Utilize the high-performance Rust backend for multi-process support. Note that sync endpoints are recommended for FastAPI integration to leverage the ThreadPool. ```python import logging from pytheus.backends import load_backend from pytheus.metrics import Counter from pytheus.exposition import generate_metrics # Enable logging to see backend initialization logging.basicConfig(level=logging.INFO) # Install: pip install pytheus-backend-rs from pytheus_backend_rs import RedisBackend # Load Rust Redis backend load_backend( backend_class=RedisBackend, backend_config={'host': '127.0.0.1', 'port': 6379} ) # Use metrics normally counter = Counter('rust_counter', 'Rust-backed counter') counter.inc() # For FastAPI async apps, use sync endpoint (runs in ThreadPool) from fastapi import FastAPI from fastapi.responses import PlainTextResponse app = FastAPI() @app.get('/metrics', response_class=PlainTextResponse) def metrics(): # Note: 'def' not 'async def' for ThreadPool execution return generate_metrics() # Single process backends also available: # from pytheus_backend_rs import SingleProcessBackend, SingleProcessAtomicBackend # load_backend(backend_class=SingleProcessAtomicBackend) ``` -------------------------------- ### Generate Metrics for Exposition Source: https://github.com/llandy3d/pytheus/blob/main/docs/quickstart.md Collect metrics from the registry and format them for Prometheus scraping. ```python from pytheus.exposition import generate_metrics metrics_data = generate_metrics() ``` -------------------------------- ### Track execution time with context manager Source: https://github.com/llandy3d/pytheus/blob/main/docs/summary.md Use the time context manager to automatically record the duration of a block of code in seconds. ```python with summary.time(): do_something() ``` -------------------------------- ### Verify default backend Source: https://github.com/llandy3d/pytheus/blob/main/docs/backend.md Check the current backend class used by a metric instance. ```python from pytheus.metrics import Counter counter = Counter('cache_hit_total', 'description') print(counter._metric_value_backend.__class__) # ``` -------------------------------- ### Redis Backend Configuration File Source: https://github.com/llandy3d/pytheus/blob/main/README.md A JSON file specifying the host and port for the Redis backend. ```json { "host": "127.0.0.1", "port": 6379 } ``` -------------------------------- ### Load Pytheus Backend Function Definition Source: https://github.com/llandy3d/pytheus/blob/main/docs/backend.md This is the function definition for loading a Pytheus backend. It should be called during initialization before creating metrics. Environment variables can automatically configure the backend class and its configuration file. ```python def load_backend( backend_class: type[Backend] | None = None, backend_config: BackendConfig | None = None, ) -> None: ``` -------------------------------- ### Track time with context manager and decorator Source: https://github.com/llandy3d/pytheus/blob/main/docs/quickstart.md Use the .time() context manager or the metric decorator to observe endpoint execution duration. ```python @app.route('/') def home(): with http_request_duration_seconds.time(): return 'hello world!' ``` ```python @app.route('/slow') @http_request_duration_seconds def slow(): time.sleep(3) return 'hello world! from slow!' ``` -------------------------------- ### Load MultiProcessRedisBackend Source: https://github.com/llandy3d/pytheus/blob/main/docs/quickstart.md Use this snippet to switch from the default SingleProcessBackend to the MultiProcessRedisBackend for storing metrics. This enables multi-process support by utilizing Redis. ```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}, ) ``` -------------------------------- ### Define metrics with required labels Source: https://github.com/llandy3d/pytheus/blob/main/docs/labels.md Initialize a Counter with required labels and create partial metric instances by pre-setting specific label values. ```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'}) ``` -------------------------------- ### Integrate FastAPI Middleware Source: https://context7.com/llandy3d/pytheus/llms.txt Uses PytheusMiddlewareASGI to automatically collect HTTP request metrics for FastAPI applications. ```python from fastapi import FastAPI from fastapi.responses import PlainTextResponse from pytheus.middleware import PytheusMiddlewareASGI from pytheus.exposition import generate_metrics app = FastAPI() # Add middleware - automatically collects: # - http_request_duration_seconds (histogram) # - http_request_size_bytes (histogram) # - http_response_size_bytes (histogram) # All with labels: method, path, status_code app.add_middleware(PytheusMiddlewareASGI) @app.get('/metrics', response_class=PlainTextResponse) async def pytheus_metrics(): """Expose metrics for Prometheus scraping.""" return generate_metrics() @app.get('/api/users') async def get_users(): return {'users': ['alice', 'bob']} @app.post('/api/users') async def create_user(name: str): return {'created': name} ``` -------------------------------- ### Using labels with dict or kwargs Source: https://github.com/llandy3d/pytheus/blob/main/docs/changelog.md Demonstrates the two ways to pass labels to a metric: using a dictionary or keyword arguments. ```python metric = Counter("counter", "desc", required_labels=["method"]) # dict approach metric.labels({"method": "POST"}) # kwargs approach metric.labels(method="POST") ``` -------------------------------- ### Observe metrics using partial labels Source: https://github.com/llandy3d/pytheus/blob/main/docs/labels.md Increment metrics by providing the remaining required labels to the partial metric instances. ```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() ``` -------------------------------- ### Track execution time with decorator Source: https://github.com/llandy3d/pytheus/blob/main/docs/summary.md Apply the summary instance as a decorator to time the execution of a function. ```python @summary def do_something(): ... ``` -------------------------------- ### Expose Metrics Endpoint Source: https://context7.com/llandy3d/pytheus/llms.txt Create a route to serve metrics in the format expected by Prometheus. ```python @app.route('/metrics') def metrics(): return Response( generate_metrics(), headers={'Content-Type': PROMETHEUS_CONTENT_TYPE} ) ``` -------------------------------- ### Instrument Routes Source: https://context7.com/llandy3d/pytheus/llms.txt Use context managers and decorators to track request duration, active requests, and increment counters within route handlers. ```python @app.route('/') def home(): with active_requests.track_inprogress(): with request_duration.labels(endpoint='/').time(): http_requests.labels(method='GET', endpoint='/', status='200').inc() return 'Hello World!' @app.route('/api/data') @request_duration.labels(endpoint='/api/data') def api_data(): with active_requests.track_inprogress(): time.sleep(0.1) # Simulate processing http_requests.labels(method='GET', endpoint='/api/data', status='200').inc() return {'data': 'sample'} ``` -------------------------------- ### Observe a metric with required labels Source: https://github.com/llandy3d/pytheus/blob/main/docs/labels.md Attempting to observe a metric without providing required labels will raise an UnobservableMetricException. ```python page_hit_total.inc() Traceback (most recent call last): File "", line 1, in File "/Users/llandy/dev/pytheus/pytheus/metrics.py", line 283, in inc self._raise_if_cannot_observe() File "/Users/llandy/dev/pytheus/pytheus/metrics.py", line 198, in _raise_if_cannot_observe raise UnobservableMetricException pytheus.exceptions.UnobservableMetricException ``` -------------------------------- ### Track Time with Histogram Context Manager Source: https://github.com/llandy3d/pytheus/blob/main/docs/histogram.md Use the `time()` context manager to automatically track the duration of a block of code in seconds. Ensure `do_something()` is defined. ```python with histogram.time(): do_something() ``` -------------------------------- ### Track Progress with Gauge Source: https://github.com/llandy3d/pytheus/blob/main/docs/gauge.md Use the track_inprogress context manager to automatically increment on entry and decrement on exit. ```python with gauge.track_inprogress(): do_something() ``` -------------------------------- ### Create Counter Without Global Registry Source: https://github.com/llandy3d/pytheus/blob/main/docs/registry.md Instantiate a Counter metric without automatically adding it to the global REGISTRY by passing `registry=None`. ```python counter = Counter('cache_hit_total', 'description', registry=None) ``` -------------------------------- ### Create Flask API Service Source: https://github.com/llandy3d/pytheus/blob/main/docs/quickstart.md 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) ``` -------------------------------- ### Backend Protocol Source: https://github.com/llandy3d/pytheus/blob/main/docs/backend.md Defines the interface required for implementing custom metric backends. ```APIDOC ## Backend Protocol ### Description Interface that all custom backends must implement to handle metric operations. ### Methods - **__init__(config, metric, histogram_bucket)**: Initializes the backend. - **inc(value)**: Increments the metric value. - **dec(value)**: Decrements the metric value. - **set(value)**: Sets the metric value. - **get()**: Returns the current metric value. - **_initialize(config)**: (Optional) Class method for one-time setup. - **_generate_samples()**: (Optional) Class method for optimized sample generation. ``` -------------------------------- ### Expose Metrics Endpoint Source: https://github.com/llandy3d/pytheus/blob/main/docs/fastapi.md Create a /metrics endpoint using the generate_metrics function to serve collected data in a format compatible with Prometheus. ```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() ``` -------------------------------- ### Configure Histogram Buckets Source: https://github.com/llandy3d/pytheus/blob/main/docs/histogram.md Customize the buckets for a Histogram metric using the `buckets` parameter. The `+Inf` bucket is automatically added. ```python customized_buckets = (0.1, 0.3, 2, 5) histogram = Histogram( name="http_request_duration_seconds", description="My description", buckets=customized_buckets, ) ``` -------------------------------- ### Register Metric with Custom Registry Source: https://github.com/llandy3d/pytheus/blob/main/docs/registry.md Create a `Counter` metric and associate it with a specific `CollectorRegistry` instance by passing the registry during metric creation. ```python my_registry = CollectorRegistry() counter = Counter('cache_hit_total', 'description', registry=my_registry) ``` -------------------------------- ### Create metrics endpoint Source: https://github.com/llandy3d/pytheus/blob/main/docs/quickstart.md Define a /metrics 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}) ``` -------------------------------- ### Configure default labels Source: https://github.com/llandy3d/pytheus/blob/main/docs/labels.md Define default values for required labels to simplify metric observation. ```python http_request_duration_seconds = Histogram( 'http_request_duration_seconds', 'documenting the metric..', required_labels=['service'], default_labels={'service': 'main_service'}, ) ``` ```python http_request_duration_seconds.observe(1) ``` ```python http_request_duration_seconds_side_service = http_request_duration_seconds.labels( {'service': 'side_service'} ) ``` -------------------------------- ### Define Custom Registry Protocol Source: https://github.com/llandy3d/pytheus/blob/main/docs/registry.md Implement the required methods (`register`, `unregister`, `collect`) to create a custom registry that adheres to the Registry Protocol. ```python class Registry(Protocol): prefix: str | None def register(self, collector: Collector) -> None: ... def unregister(self, collector: Collector) -> None: ... def collect(self) -> Iterable: ... ``` -------------------------------- ### Instrument with Histogram Source: https://context7.com/llandy3d/pytheus/llms.txt Use Histograms to track distributions of observations using configurable buckets. ```python from pytheus.metrics import Histogram # Create histogram with default buckets # Default: (.005, .01, .025, .05, .1, .25, .5, 1, 2.5, 5, 10) http_request_duration = Histogram( name='http_request_duration_seconds', description='HTTP request duration in seconds' ) # Create histogram with custom buckets response_size = Histogram( name='http_response_size_bytes', description='HTTP response size in bytes', buckets=(100, 500, 1000, 5000, 10000) ) # Observe a value http_request_duration.observe(0.4) # Time a piece of code with context manager with http_request_duration.time(): handle_request() # Use as decorator for timing @http_request_duration def process_request(): fetch_data() return response ``` -------------------------------- ### Observe a value Source: https://github.com/llandy3d/pytheus/blob/main/docs/summary.md Use the observe method to record a specific numerical value. ```python summary.observe(0.4) ``` -------------------------------- ### Generate Metrics with Custom Registry Source: https://github.com/llandy3d/pytheus/blob/main/docs/registry.md Use the `generate_metrics` function with a specific `CollectorRegistry` instance to collect and format metrics from that registry. ```python from pytheus.exposition import generate_metrics my_registry = CollectorRegistry() generate_metrics(my_registry) ``` -------------------------------- ### Add Pytheus Middleware to FastAPI Source: https://github.com/llandy3d/pytheus/blob/main/docs/fastapi.md Register the PytheusMiddlewareASGI with your FastAPI application instance to enable automatic metric collection. ```python from fastapi import FastAPI from pytheus.middleware import PytheusMiddlewareASGI app = FastAPI() app.add_middleware(PytheusMiddlewareASGI) ``` -------------------------------- ### Chain partial labels Source: https://github.com/llandy3d/pytheus/blob/main/docs/labels.md Create further specialized metric instances by chaining additional label sets. ```python cache_hit_count_total_system_a_with_origin.labels({'origin': 'function_a'}) cache_hit_count_total_system_a_with_origin.inc() ``` -------------------------------- ### Configure Redis backend expiration Source: https://github.com/llandy3d/pytheus/blob/main/docs/backend.md Set the TTL for keys in Redis using the expire_key_time configuration parameter. ```python # 5 min load_backend(MultiProcessRedisBackend, {"expire_key_time": 300}) ``` -------------------------------- ### Patch Prometheus Client for Compatibility Source: https://github.com/llandy3d/pytheus/blob/main/README.md An experimental feature to patch the prometheus_client library in-place, enabling testing of Pytheus's multiprocessing mode. ```python import prometheus_client from pytheus.experimental.compatibility import patch_client patch_client(prometheus_client) ``` -------------------------------- ### Implement Custom Collectors in Python Source: https://context7.com/llandy3d/pytheus/llms.txt Define custom collectors by inheriting from CustomCollector to aggregate metrics from external sources. Ensure metrics are created with registry=None to prevent auto-registration. ```python from pytheus.metrics import Counter, Gauge, CustomCollector from pytheus.registry import REGISTRY class DatabaseMetricsCollector(CustomCollector): def collect(self): # Create metrics without auto-registration connections = Gauge( 'db_connections_active', 'Active database connections', registry=None ) connections.set(get_active_connections()) yield connections queries = Counter( 'db_queries_total', 'Total database queries', registry=None ) queries.inc(get_query_count()) yield queries class ExternalServiceCollector(CustomCollector): def collect(self): # Fetch metrics from external API stats = fetch_external_stats() latency = Gauge('external_api_latency_seconds', 'API latency', registry=None) latency.set(stats['latency']) yield latency # Register custom collectors REGISTRY.register(DatabaseMetricsCollector()) REGISTRY.register(ExternalServiceCollector()) ``` -------------------------------- ### Define Prometheus Metrics Source: https://context7.com/llandy3d/pytheus/llms.txt Initialize Counter, Histogram, and Gauge metrics at the module level to track application state. ```python http_requests = Counter( 'http_requests_total', 'Total HTTP requests', required_labels=['method', 'endpoint', 'status'] ) request_duration = Histogram( 'http_request_duration_seconds', 'HTTP request duration', required_labels=['endpoint'], buckets=(0.01, 0.05, 0.1, 0.5, 1.0, 5.0) ) active_requests = Gauge( 'http_requests_in_progress', 'Number of requests currently being processed' ) ``` -------------------------------- ### Track Time with Gauge Source: https://github.com/llandy3d/pytheus/blob/main/docs/gauge.md Measure the duration of a code block in seconds using the time context manager. ```python with gauge.time(): do_something() ``` -------------------------------- ### Cache metric instance with labels Source: https://github.com/llandy3d/pytheus/blob/main/docs/labels.md Assign the result of the labels() method to a variable to reuse a metric instance with pre-set labels. ```python page_hit_total_with_get = page_hit_total.labels({'method': 'GET'}) ``` ```python page_hit_total_with_get.inc() ``` -------------------------------- ### Instrument with Counter Source: https://context7.com/llandy3d/pytheus/llms.txt Use Counters for monotonically increasing values or counting exceptions. ```python from pytheus.metrics import Counter # Create a counter cache_hit_total = Counter(name='cache_hit_total', description='number of cache hits') # Increment by 1 cache_hit_total.inc() # Increment by specific amount cache_hit_total.inc(7) # Count exceptions with context manager with cache_hit_total.count_exceptions(): process_request() # Increments counter if exception raised # Count specific exceptions only with cache_hit_total.count_exceptions((IndexError, ValueError)): risky_operation() # Use as decorator to count exceptions @cache_hit_total def my_func(): raise ValueError # Increments counter by 1 # Decorator with specific exceptions @cache_hit_total(exceptions=(IndexError, ValueError)) def another_func(): raise KeyError # Won't increment (not in exception list) ``` -------------------------------- ### Set Default Global Registry Source: https://github.com/llandy3d/pytheus/blob/main/docs/registry.md Replace the default global registry with a custom `CollectorRegistry` instance using the `set_registry` method on the `REGISTRY` proxy. This should be done before creating any metrics. ```python from pytheus.registry import REGISTRY, CollectorRegistry my_registry = CollectorRegistry(prefix='service_a') REGISTRY.set_registry(my_registry) ``` -------------------------------- ### Generate Metrics in FastAPI Source: https://github.com/llandy3d/pytheus/blob/main/docs/rust_backend.md Use a synchronous endpoint definition to ensure generate_metrics runs within a ThreadPool, allowing the GIL to be released. ```python @app.get('/metrics') def pytheus_metrics(): return generate_metrics() ``` -------------------------------- ### Instrument with Gauge Source: https://context7.com/llandy3d/pytheus/llms.txt Use Gauges for values that fluctuate, such as memory usage or task duration. ```python from pytheus.metrics import Gauge # Create a gauge room_temperature = Gauge('room_temperature', 'temperature in the living room') # Increment and decrement room_temperature.inc() # Increase by 1 room_temperature.inc(7) # Increase by 7 room_temperature.dec() # Decrease by 1 room_temperature.dec(3) # Decrease by 3 # Set to specific value room_temperature.set(23.5) # Set to current Unix timestamp room_temperature.set_to_current_time() # Track in-progress operations (increases on enter, decreases on exit) with room_temperature.track_inprogress(): process_long_task() # Time a piece of code (sets duration in seconds) with room_temperature.time(): do_something() # Use as decorator for timing @room_temperature def timed_function(): do_work() # Decorator for tracking in-progress @room_temperature(track_inprogress=True) def tracked_function(): process_request() ``` -------------------------------- ### Register a Custom Collector Source: https://github.com/llandy3d/pytheus/blob/main/docs/quickstart.md Manually register a collector object to the global registry. ```python REGISTRY.register(mycollector) ```