proxywhirl.retry.metrics

Retry metrics collection and aggregation.

Thread Safety:

RetryMetrics uses threading.Lock for thread-safe operation in synchronous contexts. When used in async contexts (e.g., FastAPI endpoints), methods that acquire locks should be called using asyncio.to_thread() to prevent blocking the event loop.

Usage:

Sync context (RetryExecutor, aggregation thread):

metrics.record_attempt(attempt) metrics.aggregate_hourly()

Async context (FastAPI endpoints):

summary = await asyncio.to_thread(metrics.get_summary) timeseries = await asyncio.to_thread(metrics.get_timeseries, hours=24)

Classes

CircuitBreakerEvent	Circuit breaker state change event.
HourlyAggregate	Hourly aggregated metrics.
RetryAttempt	Record of a single retry attempt.
RetryMetrics	Aggregated retry metrics collection.
RetryOutcome	Outcome of a retry attempt.

Module Contents

class proxywhirl.retry.metrics.CircuitBreakerEvent(/, **data)

Bases: pydantic.BaseModel

Circuit breaker state change event.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:

data (Any)

class proxywhirl.retry.metrics.HourlyAggregate(/, **data)

Bases: pydantic.BaseModel

Hourly aggregated metrics.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:

data (Any)

class proxywhirl.retry.metrics.RetryAttempt(/, **data)

Bases: pydantic.BaseModel

Record of a single retry attempt.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:

data (Any)

class proxywhirl.retry.metrics.RetryMetrics(**data)

Bases: pydantic.BaseModel

Aggregated retry metrics collection.

Initialize with maxlen-bounded deque.

aggregate_hourly()

Aggregate current_attempts into hourly summaries.

Return type:

None

get_by_proxy(hours=24)

Get per-proxy retry statistics.

Parameters:

hours (int)

Return type:

dict[str, dict[str, Any]]

get_summary()

Get metrics summary for API response.

Return type:

dict[str, Any]

get_timeseries(hours=24)

Get time-series data for the specified hours.

Parameters:

hours (int)

Return type:

list[dict[str, Any]]

record_attempt(attempt)

Record a retry attempt.

Parameters:

attempt (RetryAttempt)

Return type:

None

record_circuit_breaker_event(event)

Record circuit breaker state change.

Parameters:

event (CircuitBreakerEvent)

Return type:

None

class proxywhirl.retry.metrics.RetryOutcome

Bases: str, enum.Enum

Outcome of a retry attempt.

Initialize self. See help(type(self)) for accurate signature.