Module

metrics

Prometheus-compatible metrics collector.

Implements theLifecycleCollectorprotocol to track standard HTTP server metrics from lifecycle events. No external dependencies — uses internal counters that can be exported in Prometheus text format.

Metrics:

  • http_requests_total— counter by method and status
  • http_request_duration_seconds— histogram of request durations
  • http_connections_active— gauge of open connections
  • http_requests_in_flight— gauge of in-progress requests
  • http_streams_active— gauge of open streaming responses
  • http_stream_duration_seconds— histogram of completed stream lifetimes

Label stability contract (http_requests_total):

  • method— the uppercase HTTP request method as parsed (e.g. 
    ``"GET"``, ``"POST"``).  On error/early-out paths where the method

    was never parsed, the stable sentinel ``"unknown"`` is used.  The

    empty string is never emitted.
  • status— the numeric HTTP status code rendered as a string.

Thread-safe: all counters usethreading.Lockfor free-threading mode.

1Class

Classes

PrometheusCollector 4 ▼
Lifecycle collector that maintains Prometheus-compatible metrics. Thread-safe — multiple workers c…

Lifecycle collector that maintains Prometheus-compatible metrics.

Thread-safe — multiple workers can callrecord()concurrently.

Example::

collector = PrometheusCollector()
server = Server(config, app, lifecycle_collector=collector)

# Later: export metrics
text = collector.export()

Methods

record 1 ▼
Process a lifecycle event and update metrics.
def record(self, event: LifecycleEvent) -> None
Parameters
Name Type Description
event —
snapshot 0 dict[str, object] ▼
Return a snapshot of current metrics as a dict. Useful for JSON export or prog…
def snapshot(self) -> dict[str, object]

Return a snapshot of current metrics as a dict.

Useful for JSON export or programmatic access.

Returns
dict[str, object]
export 0 str ▼
Export metrics in Prometheus text exposition format.
def export(self) -> str
Returns
str String in Prometheus text format, ready to serve at ``/metrics``.
Internal Methods 1 ▼
__init__ 1 ▼
def __init__(self, *, duration_buckets: tuple[float, ...] = _DEFAULT_BUCKETS) -> None
Parameters
Name Type Description
duration_buckets — Default:_DEFAULT_BUCKETS