Metrics Utilities

Public API for performance measurement: timing, memory tracking, call statistics, and step tracking. These utilities help profile code execution and identify bottlenecks.

Tip

Pair this reference with Metrics for the feature guide.

Quick overview

  • @metrics decorator provides unified time + memory tracking with Rich output

  • @call_stats tracks call count, avg/min/max duration across multiple invocations

  • metrics_context context manager measures code blocks inline

  • Stopwatch provides manual lap timing with summary display

  • Step tracking with step=True assigns incrementing numbers for pipeline visualization

  • Configuration follows the standard priority chain: decorator args > kstlib.conf.yml > defaults

Configuration cascade

The module consults the loaded config for default values. A minimal config block:

metrics:
  colors: true
  defaults:
    time: true
    memory: true
    step: false
  thresholds:
    time_warn: 5
    time_crit: 30

Override any of these per call:

from kstlib.metrics import metrics

@metrics(memory=False, step=True)
def my_step():
    pass

Usage patterns

Basic decorator usage

from kstlib.metrics import metrics

@metrics
def process_data(items: list) -> int:
    return sum(items)

result = process_data([1, 2, 3])
# Output: [process_data (script.py:3)] | 0.001s | Peak: 64 KB

Step tracking for pipelines

from kstlib.metrics import metrics, metrics_summary, clear_metrics

clear_metrics()

@metrics(step=True)
def load():
    pass

@metrics(step=True, title="Transform")
def transform():
    pass

load()
transform()
metrics_summary()

Call statistics

from kstlib.metrics import call_stats, print_all_call_stats

@call_stats
def api_call():
    return fetch()

for _ in range(10):
    api_call()

print_all_call_stats()

Context manager

from kstlib.metrics import metrics_context

with metrics_context("Data loading") as m:
    data = load_data()

print(f"Took {m.elapsed_seconds:.3f}s, peak {m.peak_memory_formatted}")

Manual stopwatch

from kstlib.metrics import Stopwatch

sw = Stopwatch("Pipeline")
sw.start()
# ... work ...
sw.lap("Step 1")
# ... work ...
sw.lap("Step 2")
sw.stop()
sw.summary()

Module reference

Metrics and timing utilities for measuring execution performance.

This module provides a unified decorator and tools for:

  • Time Measurement: Track execution time of functions and code blocks

  • Memory Tracking: Monitor peak memory usage with tracemalloc

  • Call Statistics: Track call count, avg/min/max durations

  • Step Tracking: Numbered step tracking with summary

All behavior is config-driven with sensible defaults.

Examples

Unified metrics decorator (config defaults):

>>> from kstlib.metrics import metrics
>>> @metrics
... def slow_function():
...     return sum(range(1000000))
>>> slow_function()  
[slow_function] ⏱ 0.023s | 🧠 Peak: 128 KB

Step tracking for pipelines:

>>> from kstlib.metrics import metrics, metrics_summary, clear_metrics
>>> @metrics(step=True)
... def load_data():
...     pass
>>> @metrics(step=True, title="Process records")
... def process():
...     pass
>>> load_data()  
[STEP 1] load_data ⏱ 0.001s
>>> process()  
[STEP 2] Process records ⏱ 0.002s
>>> metrics_summary()  
# Displays summary table with all steps

Context manager usage:

>>> from kstlib.metrics import metrics_context
>>> with metrics_context("Data loading"):  
...     data = load_large_file()
[Data loading] ⏱ 2.34s | 🧠 Peak: 512 MB

Manual stopwatch:

>>> from kstlib.metrics import Stopwatch
>>> sw = Stopwatch("Pipeline")
>>> sw.start()  
>>> # ... work ...
>>> sw.lap("Step 1")  
>>> sw.stop()  
>>> sw.summary()
class kstlib.metrics.CallStats(name, call_count=0, total_time=0.0, min_time=inf, max_time=0.0, _lock=<factory>)[source]

Bases: object

Statistics for tracked function calls.

Examples

>>> stats = CallStats("my_func")
>>> stats.record(0.5)
>>> stats.record(1.0)
>>> stats.call_count
2
>>> stats.avg_time
0.75
name: str
call_count: int = 0
total_time: float = 0.0
min_time: float = inf
max_time: float = 0.0
record(self, elapsed: 'float') → 'None' -> None[source]

Record a call duration.

property avg_time: float

Return average call duration.

reset(self) → 'None' -> None[source]

Reset all statistics.

__str__(self) → 'str' -> str[source]

Return human-readable summary with Rich colors.

__init__(self, name: 'str', call_count: 'int' = 0, total_time: 'float' = 0.0, min_time: 'float' = inf, max_time: 'float' = 0.0, _lock: 'threading.Lock' = <factory>) → None -> None
exception kstlib.metrics.MetricsError[source]

Bases: KstlibError

Base exception for metrics-related errors.

Examples

>>> raise MetricsError("Something went wrong")
Traceback (most recent call last):
    ...
kstlib.metrics.exceptions.MetricsError: Something went wrong
class kstlib.metrics.MetricsRecord(number, title, elapsed_seconds=0.0, peak_memory_bytes=None, function='', module='', file='', line=0)[source]

Bases: object

Record of a metrics measurement.

number

Step number (if step=True).

Type:

int | None

title

Display title.

Type:

str

elapsed_seconds

Execution time.

Type:

float

peak_memory_bytes

Peak memory usage.

Type:

int | None

function

Function name.

Type:

str

module

Module name.

Type:

str

file

Source file.

Type:

str

line

Source line.

Type:

int

number: int | None
title: str
elapsed_seconds: float = 0.0
peak_memory_bytes: int | None = None
function: str = ''
module: str = ''
file: str = ''
line: int = 0
property elapsed_formatted: str

Return formatted elapsed time.

property peak_memory_formatted: str | None

Return formatted peak memory.

__init__(self, number: 'int | None', title: 'str', elapsed_seconds: 'float' = 0.0, peak_memory_bytes: 'int | None' = None, function: 'str' = '', module: 'str' = '', file: 'str' = '', line: 'int' = 0) → None -> None
class kstlib.metrics.Stopwatch(name='Stopwatch', _start_time=None, _lap_start=None, _laps=<factory>, _stopped=False, _lock=<factory>)[source]

Bases: object

Manual stopwatch for timing code sections.

For cases where you need explicit start/stop/lap control.

Examples

>>> sw = Stopwatch("Pipeline")
>>> _ = sw.start()
>>> # ... work ...
>>> sw.lap("Step 1")  
>>> _ = sw.stop()
>>> sw.summary()
name: str = 'Stopwatch'
start(self) → 'Stopwatch' -> Stopwatch[source]

Start the stopwatch.

lap(self, name: 'str', *, print_result: 'bool' = True, track_memory: 'bool' = False) → 'float' -> float[source]

Record a lap time.

stop(self) → 'float' -> float[source]

Stop the stopwatch.

property total_elapsed: float

Get total elapsed time.

property laps: list[tuple[str, float, int | None]]

Get all recorded laps.

reset(self) → 'None' -> None[source]

Reset the stopwatch.

summary(self, *, show_percentages: 'bool' = True) → 'None' -> None[source]

Print a summary of all laps.

__init__(self, name: 'str' = 'Stopwatch', _start_time: 'float | None' = None, _lap_start: 'float | None' = None, _laps: 'list[tuple[str, float, int | None]]' = <factory>, _stopped: 'bool' = False, _lock: 'threading.RLock' = <factory>) → None -> None
kstlib.metrics.call_stats(func: 'Callable[P, R] | None' = None, /, *, name: 'str | None' = None, print_on_call: 'bool' = False) → 'Callable[P, R] | Callable[[Callable[P, R]], Callable[P, R]]' -> Callable[P, R] | Callable[[Callable[P, R]], Callable[P, R]][source]

Track call statistics for the decorated function.

Tracks call count, total time, min, max, and average duration. Use get_call_stats() or print_all_call_stats() to access results.

Examples

>>> @call_stats
... def api_call():
...     pass
>>> api_call()
>>> api_call()
>>> stats = get_call_stats("api_call")
>>> stats.call_count
2
kstlib.metrics.clear_metrics() → 'None' -> None[source]

Clear all recorded metrics and reset step counter.

Examples

>>> clear_metrics()
kstlib.metrics.get_all_call_stats() → 'dict[str, CallStats]' -> dict[str, CallStats][source]

Get all tracked call statistics.

kstlib.metrics.get_call_stats(func_name: 'str') → 'CallStats | None' -> CallStats | None[source]

Get call statistics for a tracked function.

kstlib.metrics.get_metrics() → 'list[MetricsRecord]' -> list[MetricsRecord][source]

Get all recorded metrics.

Returns:

List of MetricsRecord objects.

Return type:

list[MetricsRecord]

Examples

>>> records = get_metrics()
>>> isinstance(records, list)
True
kstlib.metrics.metrics(func_or_title: 'Callable[P, R] | str | None' = None, /, *, title: 'str | None' = None, time: 'bool | None' = None, memory: 'bool | None' = None, step: 'bool | None' = None, print_result: 'bool' = True) → 'Callable[P, R] | Callable[[Callable[P, R]], Callable[P, R]]' -> Callable[P, R] | Callable[[Callable[P, R]], Callable[P, R]][source]

Unified decorator for timing, memory tracking, and step counting.

Config-driven with sensible defaults. All options can be overridden.

Parameters:

  • func_or_title (Callable[P, R] | str | None) – Function to decorate or custom title string.

  • title (str | None) – Custom title (alternative to positional).

  • time (bool | None) – Track execution time (default from config, typically True).

  • memory (bool | None) – Track peak memory (default from config, typically True).

  • step (bool | None) – Enable step numbering (default from config, typically False).

  • print_result (bool) – Whether to print result to stderr.

Returns:

Decorated function.

Return type:

Callable[P, R] | Callable[[Callable[P, R]], Callable[P, R]]

Examples

Default behavior (time + memory from config):

>>> @metrics
... def process():
...     return sum(range(1000))
>>> process()  
[process] ⏱ 0.001s | 🧠 Peak: 64 KB

Time only:

>>> @metrics(memory=False)
... def quick():
...     pass

With step numbering:

>>> @metrics(step=True)
... def step1():
...     pass
>>> step1()  
[STEP 1] step1 ⏱ 0.001s | 🧠 Peak: 32 KB

Custom title:

>>> @metrics("Loading configuration")
... def load_config():
...     pass
kstlib.metrics.metrics_context(title: 'str', *, time: 'bool | None' = None, memory: 'bool | None' = None, step: 'bool | None' = None, print_result: 'bool' = True) → 'Generator[MetricsRecord, None, None]' -> Generator[MetricsRecord, None, None][source]

Context manager for metrics tracking.

Parameters:

  • title (str) – Title for this measurement.

  • time (bool | None) – Track execution time.

  • memory (bool | None) – Track peak memory.

  • step (bool | None) – Enable step numbering.

  • print_result (bool) – Whether to print result.

Yields:

MetricsRecord being tracked.

Examples

>>> with metrics_context("Loading data") as m:  
...     data = load_file()
[Loading data] ⏱ 1.23s | 🧠 Peak: 256 MB
kstlib.metrics.metrics_summary(*, show_percentages: 'bool' = True, style: 'str' = 'table') → 'None' -> None[source]

Print summary of all recorded metrics (step=True records).

Uses kstlib.ui.tables for pretty output.

Parameters:

  • show_percentages (bool) – Whether to show percentage of total time.

  • style (str) – Output style - “table” (Rich table) or “simple” (plain text).

Examples

>>> metrics_summary()
kstlib.metrics.print_all_call_stats() → 'None' -> None[source]

Print all tracked call statistics to stderr.

kstlib.metrics.reset_all_call_stats() → 'None' -> None[source]

Reset all tracked call statistics.

Exceptions

Exceptions for the metrics module.

exception kstlib.metrics.exceptions.MetricsError[source]

Bases: KstlibError

Base exception for metrics-related errors.

Examples

>>> raise MetricsError("Something went wrong")
Traceback (most recent call last):
    ...
kstlib.metrics.exceptions.MetricsError: Something went wrong