Metrics

metrics ¶

In-process call metrics: count_calls tracks calls, errors, concurrency and latency.

CallStats ¶

CallStats(window: int)

Live statistics of a function wrapped by count_calls.

Every attribute is read under a lock at access time, so f.stats.calls always reflects the present. Latency figures (mean, p50, p95, p99) cover the last window finished calls, successful or not, in seconds; they are 0.0 before the first call finishes. calls and errors count every call since the last reset, not just the window.

Attributes:

Name	Type	Description
`calls`	`int`	Calls that have finished (returned or raised).
`errors`	`int`	Calls that raised an exception.
`in_flight`	`int`	Calls running right now.
`last_duration`	`float`	Seconds the most recently finished call took.
`mean`	`float`	Mean duration over the window.
`p50`	`float`	Median duration over the window (nearest rank).
`p95`	`float`	95th percentile duration over the window.
`p99`	`float`	99th percentile duration over the window.

Counted ¶

Bases: Protocol[P, R_co]

A function wrapped by count_calls (callable plus stats and reset()).

Use it to annotate decorated functions, e.g. Counted[[int], str]. The __get__ overloads let type checkers bind self on instance access and cls under a @classmethod stacked above, with the same typing limitations as Memoized (see python_wrap.caching). The statistics belong to the function, so they are shared by every instance of a class.

reset ¶

reset() -> None

Zero the counters and clear the duration window.

count_calls ¶

count_calls(func: Callable[P, R]) -> Counted[P, R]

count_calls(func: None = None, /, *, window: int = 1000) -> Callable[[Callable[P, R]], Counted[P, R]]

count_calls(func: Callable[..., Any] | None = None, /, *, window: int = 1000) -> Any

Track how often a function runs, fails, and how long it takes.

The wrapper exposes .stats (a live CallStats: calls, errors, in_flight, last_duration, mean, p50, p95, p99) and .reset(). Sync and async def functions are supported (async calls are timed across the whole awaited coroutine), as are methods; statistics are per function, not per instance. A call that raises counts as an error and its exception propagates. Cancellation counts as an error too. Updates are thread-safe.

Parameters:

Name	Type	Description	Default
`func`	`Callable[..., Any] \| None`	The function, when used as a bare `@count_calls`.	`None`
`window`	`int`	How many of the most recent durations feed `mean` and the percentiles.	`1000`

Returns:

Type	Description
`Any`	The wrapped function, a `Counted`.

Raises:

Type	Description
`ValueError`	If `window` is less than 1.

Example

@count_calls(window=500) ... def work() -> int: ... return 1 work() 1 work.stats.calls 1