class documentation

class AuthMetricsRecorder: (source)

Constructor: AuthMetricsRecorder(metrics, provider)

View In Hierarchy

Small mutable auth metrics recorder with a fixed provider label.

Method __init__ Undocumented
Method cache_hit Emit a cache hit event.
Method cache_invalidate Emit a cache invalidation event.
Method cache_miss Emit a cache miss event.
Method cache_refresh Emit a cache refresh event.
Method cache_store Emit a cache store event.
Method set_metrics Replace callbacks, sharing callback state with recorders when given.
Method token_acquire Emit a token acquisition event.
Method token_acquire_from_start Emit token acquisition duration and optional token lifetime.
Method token_lifetime Emit token lifetime when the token has an aware expiration timestamp.
Method token_refresh Emit a token refresh event.
Instance Variable provider Undocumented
Instance Variable _cell Undocumented
def __init__(self, metrics: AuthMetricsLike, provider: str): (source)

Undocumented

def cache_hit(self): (source)

Emit a cache hit event.

def cache_invalidate(self): (source)

Emit a cache invalidation event.

def cache_miss(self, result: MetricResult): (source)

Emit a cache miss event.

def cache_refresh(self, result: MetricResult): (source)

Emit a cache refresh event.

def cache_store(self, result: MetricResult): (source)

Emit a cache store event.

def set_metrics(self, metrics: AuthMetricsLike): (source)

Replace callbacks, sharing callback state with recorders when given.

def token_acquire(self, result: MetricResult, duration_seconds: float, attempt: int): (source)

Emit a token acquisition event.

def token_acquire_from_start(self, result: MetricResult, start: float, attempt: int, token: object | None = None): (source)

Emit token acquisition duration and optional token lifetime.

start must come from metric_start. When token is provided for a successful acquisition, its lifetime is emitted too.

def token_lifetime(self, token: object): (source)

Emit token lifetime when the token has an aware expiration timestamp.

def token_refresh(self, result: MetricResult, duration_seconds: float, background: bool = True): (source)

Emit a token refresh event.

provider = (source)

Undocumented