Skip to content

metrics

Here's the reference for the Prometheus metrics registry and endpoint handler.

You can import them directly from fastapi_toolsets.metrics:

from fastapi_toolsets.metrics import Metric, MetricsRegistry, init_metrics

fastapi_toolsets.metrics.registry.Metric dataclass

A metric definition with metadata.

fastapi_toolsets.metrics.registry.MetricsRegistry

Registry for managing Prometheus metric providers and collectors.

Example 
from prometheus_client import Counter, Gauge
from fastapi_toolsets.metrics import MetricsRegistry

metrics = MetricsRegistry()

@metrics.register
def http_requests():
    return Counter("http_requests_total", "Total HTTP requests", ["method", "status"])

@metrics.register(name="db_pool")
def database_pool_size():
    return Gauge("db_pool_size", "Database connection pool size")

@metrics.register(collect=True)
def collect_queue_depth(gauge=Gauge("queue_depth", "Current queue depth")):
    gauge.set(get_current_queue_depth())

get_all()

Get all registered metric definitions.

get_collectors()

Get collectors (called on each scrape).

get_providers()

Get metric providers (called once at init).

include_registry(registry)

Include another :class:MetricsRegistry into this one.

Parameters:

Name Type Description Default
registry MetricsRegistry

The registry to merge in.

 required

Raises:

Type Description
ValueError

If a metric name already exists in the current registry.
Example 
main = MetricsRegistry()
sub = MetricsRegistry()

@sub.register
def sub_metric():
    return Counter("sub_total", "Sub counter")

main.include_registry(sub)

register(func=None, *, name=None, collect=False)

Register a metric provider or collector function.

Can be used as a decorator with or without arguments.

Parameters:

Name Type Description Default
func Callable[..., Any] | None

The metric function to register.

 None
name str | None

Metric name (defaults to function name).

 None
collect bool

If True, the function is called on every scrape. If False (default), called once at init time.

 False
Example 
@metrics.register
def my_counter():
    return Counter("my_counter", "A counter")

@metrics.register(collect=True, name="queue")
def collect_queue_depth():
    gauge.set(compute_depth())

fastapi_toolsets.metrics.handler.init_metrics(app, registry, *, path='/metrics')

Register a Prometheus /metrics endpoint on a FastAPI app.

Parameters:

Name Type Description Default
app FastAPI

FastAPI application instance.

 required
registry MetricsRegistry

A :class:MetricsRegistry containing providers and collectors.

 required
path str

URL path for the metrics endpoint (default /metrics).

 '/metrics'

Returns:

Type Description
FastAPI

The same FastAPI instance (for chaining).
Example 
from fastapi import FastAPI
from fastapi_toolsets.metrics import MetricsRegistry, init_metrics

metrics = MetricsRegistry()
app = FastAPI()
init_metrics(app, registry=metrics)