pyphoenix.metrics

Metrics and monitoring system for PyPhoenix.

Functions

`get_metrics_registry`()	Get the global metrics registry.
`get_phoenix_metrics`()	Get the Phoenix metrics collector.

Classes

`Counter`(name[, description])	Thread-safe counter metric.
`Gauge`(name[, description])	Thread-safe gauge metric.
`Histogram`(name[, description, buckets])	Thread-safe histogram metric.
`MetricValue`(value, timestamp[, labels])	A metric value with timestamp.
`MetricsCollector`(registry[, collect_interval])	Background metrics collector.
`MetricsRegistry`()	Registry for managing metrics.
`PhoenixMetrics`(registry)	Phoenix-specific metrics collector.

class pyphoenix.metrics.MetricValue(value, timestamp, labels=<factory>)[source]

A metric value with timestamp.

Parameters:

value (float)
timestamp (float)
labels (dict[str, str])

value: float

timestamp: float

labels: dict[str, str]

__init__(value, timestamp, labels=<factory>)

Parameters:

value (float)
timestamp (float)
labels (dict[str, str])

class pyphoenix.metrics.Counter(name, description='')[source]

Thread-safe counter metric.

Parameters:

name (str)
description (str)

__init__(name, description='')[source]

Parameters:

name (str)
description (str)

async inc(amount=1.0, labels=None)[source]

Increment the counter.

Parameters:

amount (float)
labels (Optional[dict[str, str]])

Return type:

None

async get()[source]

Get the current counter value.

Return type:: float

async reset()[source]

Reset the counter to zero.

Return type:: None

class pyphoenix.metrics.Gauge(name, description='')[source]

Thread-safe gauge metric.

Parameters:

name (str)
description (str)

__init__(name, description='')[source]

Parameters:

name (str)
description (str)

async set(value, labels=None)[source]

Set the gauge value.

Parameters:

value (float)
labels (Optional[dict[str, str]])

Return type:

None

async inc(amount=1.0, labels=None)[source]

Increment the gauge.

Parameters:

amount (float)
labels (Optional[dict[str, str]])

Return type:

None

async dec(amount=1.0, labels=None)[source]

Decrement the gauge.

Parameters:

amount (float)
labels (Optional[dict[str, str]])

Return type:

None

async get()[source]

Get the current gauge value.

Return type:: float

class pyphoenix.metrics.Histogram(name, description='', buckets=None)[source]

Thread-safe histogram metric.

Parameters:

name (str)
description (str)
buckets (Optional[list[float]])

__init__(name, description='', buckets=None)[source]

Parameters:

name (str)
description (str)
buckets (Optional[list[float]])

async observe(value, labels=None)[source]

Observe a value.

Parameters:

value (float)
labels (Optional[dict[str, str]])

Return type:

None

async get_stats()[source]

Get histogram statistics.

Return type:: dict[str, Any]

class pyphoenix.metrics.MetricsRegistry[source]

Registry for managing metrics.

__init__()[source]

async counter(name, description='')[source]

Get or create a counter metric.

Parameters:

name (str)
description (str)

Return type:

Counter

async gauge(name, description='')[source]

Get or create a gauge metric.

Parameters:

name (str)
description (str)

Return type:

Gauge

async histogram(name, description='', buckets=None)[source]

Get or create a histogram metric.

Parameters:

name (str)
description (str)
buckets (Optional[list[float]])

Return type:

Histogram

async get_all_metrics()[source]

Get all metrics and their values.

Return type:: dict[str, Any]

async clear()[source]

Clear all metrics.

Return type:: None

class pyphoenix.metrics.PhoenixMetrics(registry)[source]

Phoenix-specific metrics collector.

Parameters:: registry (MetricsRegistry)

__init__(registry)[source]

Parameters:: registry (MetricsRegistry)

async record_connection(transport='websocket')[source]

Record a new connection.

Parameters:: transport (str)
Return type:: None

async record_disconnection(transport='websocket')[source]

Record a disconnection.

Parameters:: transport (str)
Return type:: None

async record_channel_join(topic, success=True)[source]

Record a channel join attempt.

Parameters:

topic (str)
success (bool)

Return type:

None

async record_channel_leave(topic)[source]

Record a channel leave.

Parameters:: topic (str)
Return type:: None

async record_message_sent(topic, event)[source]

Record a message sent.

Parameters:

topic (str)
event (str)

Return type:

None

async record_message_received(topic, event)[source]

Record a message received.

Parameters:

topic (str)
event (str)

Return type:

None

async record_message_processing_time(topic, event, duration)[source]

Record message processing time.

Parameters:

topic (str)
event (str)
duration (float)

Return type:

None

async record_presence_track(topic)[source]

Record presence tracking.

Parameters:: topic (str)
Return type:: None

async record_presence_untrack(topic)[source]

Record presence untracking.

Parameters:: topic (str)
Return type:: None

async record_pubsub_publish(topic, subscribers)[source]

Record PubSub message publication.

Parameters:

topic (str)
subscribers (int)

Return type:

None

async record_error(error_type, component)[source]

Record an error.

Parameters:

error_type (str)
component (str)

Return type:

None

class pyphoenix.metrics.MetricsCollector(registry, collect_interval=60.0)[source]

Background metrics collector.

Parameters:

registry (MetricsRegistry)
collect_interval (float)

__init__(registry, collect_interval=60.0)[source]

Parameters:

registry (MetricsRegistry)
collect_interval (float)

add_collector(collector)[source]

Add a metrics collector function.

Parameters:: collector (Callable[[], Any])
Return type:: None

async start()[source]

Start the metrics collection loop.

Return type:: None

async stop()[source]

Stop the metrics collection loop.

Return type:: None

async pyphoenix.metrics.get_metrics_registry()[source]

Get the global metrics registry.

Return type:: MetricsRegistry

async pyphoenix.metrics.get_phoenix_metrics()[source]

Get the Phoenix metrics collector.

Return type:: PhoenixMetrics