API Reference¶

Configuration¶

User-facing configuration models for rampa.

All scenario, executor, threshold, and option configuration is defined here as pydantic models with validation. These models are the user’s primary interface for configuring load tests.

>>> import rampa.config

rampa.config.parse_duration(value) ¶ function function rampa.config.parse_duration(value) ¶

Parse a human-readable duration string into a timedelta.

Supports h, m, s, and ms components. At least one component must be present.

Parameters:

value (str | datetime.timedelta) – Duration string (e.g. "30s", "1m30s", "2h") or an existing timedelta.

Returns:

Parsed duration.

Return type:

datetime.timedelta

Raises:

ValueError – If the string cannot be parsed.
>>> parse_duration("30s") –
datetime.timedelta(seconds=30) –
>>> parse_duration("1m30s") –
datetime.timedelta(seconds=90) –
>>> parse_duration("2h") –
datetime.timedelta(seconds=7200) –
>>> parse_duration("500ms") –
datetime.timedelta(microseconds=500000) –
>>> parse_duration("1h30m15s") –
datetime.timedelta(seconds=5415) –
>>> parse_duration(datetime.timedelta(seconds=10)) –
datetime.timedelta(seconds=10) –

rampa.config.Duration ¶ data data rampa.config.Duration ¶: A timedelta that can be constructed from a human-readable string.

alias of Annotated[timedelta, BeforeValidator(func=parse_duration, json_schema_input_type=PydanticUndefined)]

class rampa.config.Stage ¶ class class class rampa.config.Stage ¶

Bases: BaseModel

A single ramp stage with a target VU or rate count and duration.

>>> import datetime
>>> Stage(
...     duration=datetime.timedelta(seconds=30), target=100,
... ).target
100
>>> Stage(
...     duration=datetime.timedelta(minutes=1), target=0,
... ).duration
datetime.timedelta(seconds=60)

class rampa.config.ScenarioConfig ¶ class class class rampa.config.ScenarioConfig ¶

Bases: BaseModel

Configuration for a single test scenario.

Each scenario maps to an executor that controls how iterations are scheduled. Executor-specific fields are validated by the executor, not by this model.

>>> import datetime
>>> cfg = ScenarioConfig(
...     executor="constant-vus",
...     vus=10,
...     duration=datetime.timedelta(seconds=30),
... )
>>> cfg.executor
'constant-vus'
>>> cfg.vus
10

class rampa.config.Options ¶ class class class rampa.config.Options ¶

Bases: BaseModel

Shortcut execution options.

These provide a simpler alternative to explicit scenarios for common cases. When both shortcuts and scenarios are provided, validation rejects the config.

>>> import datetime
>>> Options(
...     vus=10, duration=datetime.timedelta(seconds=30),
... ).vus
10

class rampa.config.Config ¶ class class class rampa.config.Config ¶

Bases: BaseModel

Top-level test configuration.

Either scenarios or shortcut options (vus, duration, etc.) may be provided, but not both.

>>> import datetime
>>> cfg = Config(
...     scenarios={"smoke": ScenarioConfig(
...         executor="constant-vus",
...         vus=1,
...         duration=datetime.timedelta(seconds=10),
...     )},
... )
>>> "smoke" in cfg.scenarios
True
>>> cfg.scenarios["smoke"].executor
'constant-vus'

Engine¶

Headless engine and run controller for rampa.

The engine owns execution state and cleanup. Frontends own presentation, JSON shape, and process exit behavior. This separation enables CLI, TUI, MCP, pytest, and GitHub Action integration without reaching into internals.

>>> import rampa.engine

class rampa.engine.EngineOptions ¶ class class class rampa.engine.EngineOptions ¶

Bases: object

Configuration for engine behavior.

>>> opts = EngineOptions()
>>> opts.metric_flush_interval
0.05

class rampa.engine.RunController ¶ class class class rampa.engine.RunController ¶

Bases: object

Control surface for a running test.

Returned by Engine.start(). Provides wait(), stop(), snapshot(), and events() for frontend consumption.

>>> import rampa.engine

class rampa.engine.Engine ¶ class class class rampa.engine.Engine ¶

Bases: object

Headless load testing engine.

Constructs per-run state, starts scenarios, and returns a controller.

Parameters:

plan (TestPlan) – Resolved test plan from the loader.
options (EngineOptions | None) – Optional engine configuration.
rampa.engine (>>> import)

_run(run_id, registry, sample_queue, metric_engine, abort_event, bus, installed_signals=None, pause_controller=None) ¶ async method async method _run(run_id, registry, sample_queue, metric_engine, abort_event, bus, installed_signals=None, pause_controller=None) ¶

Execute the full lifecycle: setup → execute → teardown.

Parameters:

run_id (str)
registry (MetricRegistry)
sample_queue (SimpleQueue[Sample | None])
metric_engine (MetricEngine)
abort_event (Event)
bus (EventBus)
installed_signals (list[int] | None)
pause_controller (PauseController | None)

Return type:

RunResult

Event Bus¶

Broadcast event bus for multi-consumer event delivery.

The EventBus fans out EngineEvent objects to any number of subscribers. Each subscriber gets its own bounded asyncio.Queue. Slow consumers are isolated — a full subscriber queue causes that subscriber’s events to be dropped without blocking the engine.

>>> import rampa.bus

class rampa.bus.EventBus ¶ class class class rampa.bus.EventBus ¶

Bases: object

Broadcast pub-sub bus for engine lifecycle events.

Thread-safety: the subscriber list is protected by a threading.Lock. Publishing from the event-loop thread uses publish(). Publishing from a background thread (e.g. MetricEngine) uses publish_threadsafe().

Parameters:

loop (asyncio.AbstractEventLoop) – The event loop that owns subscriber queues.
maxsize (int) – Maximum queue depth per subscriber before events are dropped.
asyncio (>>> import)
asyncio.new_event_loop() (>>> loop =)
EventBus(loop (>>> bus =)
maxsize=100)
bus.subscribe() (>>> q =)
bus.publish(None) (>>>)
None (>>> loop.run_until_complete(q.get()) is)
True
bus.unsubscribe(q) (>>>)
loop.close() (>>>)

Events¶

Typed engine events and run status for rampa.

Events are observational — engine correctness does not depend on an event consumer being active. Frontends subscribe to events for live updates.

>>> import rampa.events

class rampa.events.RunStatus ¶ class class class rampa.events.RunStatus ¶

Bases: StrEnum

Final status of a completed test run.

>>> RunStatus.PASSED.value
'passed'
>>> RunStatus.THRESHOLD_FAILED.value
'threshold_failed'

rampa.events._make_run_id() ¶ function function rampa.events._make_run_id() ¶

Return type:: str

class rampa.events.EngineEvent ¶ class class class rampa.events.EngineEvent ¶

Bases: object

Base class for all engine events.

>>> e = EngineEvent(run_id="abc", timestamp_ns=0)
>>> e.run_id
'abc'

class rampa.events.PhaseEvent ¶ class class class rampa.events.PhaseEvent ¶

Bases: EngineEvent

Lifecycle phase transition.

>>> e = PhaseEvent(run_id="abc", timestamp_ns=0, phase="setup")
>>> e.phase
'setup'

class rampa.events.PauseEvent ¶ class class class rampa.events.PauseEvent ¶

Bases: EngineEvent

Emitted when execution is paused.

>>> e = PauseEvent(run_id="x", timestamp_ns=0)
>>> e.run_id
'x'

class rampa.events.ResumeEvent ¶ class class class rampa.events.ResumeEvent ¶

Bases: EngineEvent

Emitted when execution resumes after a pause.

>>> e = ResumeEvent(run_id="x", timestamp_ns=0, paused_seconds=1.5)
>>> e.paused_seconds
1.5

class rampa.events.SnapshotEvent ¶ class class class rampa.events.SnapshotEvent ¶

Bases: EngineEvent

Periodic metric snapshot.

>>> from rampa.metrics import MetricSnapshot
>>> snap = MetricSnapshot(timestamp=0, duration=1.0, values={})
>>> e = SnapshotEvent(run_id="x", timestamp_ns=0, snapshot=snap)
>>> e.snapshot.duration
1.0

class rampa.events.ThresholdEvent ¶ class class class rampa.events.ThresholdEvent ¶

Bases: EngineEvent

Threshold evaluation results.

>>> e = ThresholdEvent(run_id="x", timestamp_ns=0, results=[])
>>> len(e.results)
0

class rampa.events.LiveThresholdEvent ¶ class class class rampa.events.LiveThresholdEvent ¶

Bases: EngineEvent

Mid-run threshold evaluation result.

Emitted periodically during execution by the MetricEngine.

>>> e = LiveThresholdEvent(
...     run_id="x", timestamp_ns=0, results=[], will_abort=False,
... )
>>> e.will_abort
False

class rampa.events.RunResult ¶ class class class rampa.events.RunResult ¶

Bases: object

Result of a completed test run.

>>> r = RunResult(
...     run_id="abc",
...     status=RunStatus.PASSED,
...     snapshot=None,
...     threshold_results=[],
... )
>>> r.status
<RunStatus.PASSED: 'passed'>

rampa.events.serialize_event(event) ¶ function function rampa.events.serialize_event(event) ¶

Serialize an engine event to a JSON-compatible dict.

Adds a type key with the event class name. All dataclass fields are included via dataclasses.asdict().

Parameters:

event (EngineEvent) – The event to serialize.

Returns:

dict[str, Any] – JSON-serializable dictionary.
>>> e = PhaseEvent(run_id=”abc”, timestamp_ns=0, phase=”setup”)
>>> d = serialize_event(e)
>>> d[“type”]
’PhaseEvent’
>>> d[“phase”]
’setup’

Return type:

dict[str, Any]

Worker¶

Worker object passed to user scenario functions.

The Worker provides a scoped API surface for one iteration. It holds references to the sample queue for metric emission and provides check and custom metric helpers.

>>> import rampa.worker

class rampa.worker.ExecutionInfo ¶ class class class rampa.worker.ExecutionInfo ¶

Bases: object

Execution context for the current worker iteration.

>>> info = ExecutionInfo(worker_id=1, scenario="load", iteration=0)
>>> info.worker_id
1

class rampa.worker.Worker ¶ class class class rampa.worker.Worker ¶

Bases: object

Scoped API surface for user code within a single iteration.

Workers are created per-iteration by the executor. They provide check(), counter(), gauge(), and trend() methods that emit metric samples to the engine’s sample queue.

Parameters:

sample_queue (queue.SimpleQueue[Sample | None]) – Queue for emitting metric samples.
execution (ExecutionInfo) – Current execution context.
setup_data (Any) – Data returned from the setup() function.

Examples

>>> import queue as q
>>> sq: q.SimpleQueue[Sample | None] = q.SimpleQueue()
>>> w = Worker(
...     sample_queue=sq,
...     execution=ExecutionInfo(worker_id=1, scenario="s", iteration=0),
... )
>>> w.execution.worker_id
1

property http: HttpClient ¶ property property property http: HttpClient ¶

Lazily-initialized HTTP client with automatic metric emission.

The client is created on first access, so non-HTTP scenarios never allocate an aiohttp session.

>>> import queue as q
>>> sq: q.SimpleQueue[Sample | None] = q.SimpleQueue()
>>> w = Worker(
...     sample_queue=sq,
...     execution=ExecutionInfo(
...         worker_id=1, scenario="s", iteration=0,
...     ),
... )
>>> w._http is None
True
>>> client = w.http
>>> w._http is not None
True

check(value, conditions) ¶ method method check(value, conditions) ¶

Evaluate named conditions and emit check metric samples.

Each condition emits one sample on the checks metric: value 1.0 for pass, 0.0 for fail. The check tag carries the condition name.

Parameters:

value (Any) – The value to test (typically an HTTP response).
conditions (dict[str, Callable[[Any], bool]]) – Mapping of check name to predicate function.

Returns:

bool – True if all conditions passed.
>>> import queue as q
>>> sq (q.SimpleQueue[Sample | None] = q.SimpleQueue())
>>> w = Worker(
… sample_queue=sq,
… execution=ExecutionInfo(
… worker_id=1, scenario=”s”, iteration=0,
… ),
… )
>>> w.check(200, {“is 200” (lambda v: v == 200}))
True
>>> s = sq.get_nowait()
>>> s.metric
’checks’
>>> s.value
1.0
>>> s.tags[“check”]
’is 200’

Return type:

bool

counter(name, value=1.0, tags=None) ¶ method method counter(name, value=1.0, tags=None) ¶

Emit a custom counter metric sample.

Parameters:

name (str) – Metric name.
value (float) – Value to add (default 1.0).
tags (dict[str, str] | None) – Optional tags.

Return type:

None

Examples

>>> import queue as q
>>> sq: q.SimpleQueue[Sample | None] = q.SimpleQueue()
>>> w = Worker(
...     sample_queue=sq,
...     execution=ExecutionInfo(
...         worker_id=1, scenario="s", iteration=0,
...     ),
... )
>>> w.counter("my_counter", 5.0)
>>> sq.get_nowait().value
5.0

gauge(name, value, tags=None) ¶ method method gauge(name, value, tags=None) ¶

Emit a custom gauge metric sample.

Parameters:

name (str) – Metric name.
value (float) – Current value.
tags (dict[str, str] | None) – Optional tags.

Return type:

None

Examples

>>> import queue as q
>>> sq: q.SimpleQueue[Sample | None] = q.SimpleQueue()
>>> w = Worker(
...     sample_queue=sq,
...     execution=ExecutionInfo(
...         worker_id=1, scenario="s", iteration=0,
...     ),
... )
>>> w.gauge("queue_depth", 42.0)
>>> sq.get_nowait().value
42.0

trend(name, value, tags=None) ¶ method method trend(name, value, tags=None) ¶

Emit a custom trend metric sample.

Parameters:

name (str) – Metric name.
value (float) – Observed value.
tags (dict[str, str] | None) – Optional tags.

Return type:

None

Examples

>>> import queue as q
>>> sq: q.SimpleQueue[Sample | None] = q.SimpleQueue()
>>> w = Worker(
...     sample_queue=sq,
...     execution=ExecutionInfo(
...         worker_id=1, scenario="s", iteration=0,
...     ),
... )
>>> w.trend("latency", 123.4)
>>> sq.get_nowait().value
123.4

HTTP Client¶

HTTP client with automatic metric emission for rampa.

Wraps aiohttp.ClientSession and emits timing metrics for every request: http_reqs, http_req_duration, http_req_failed, data transfer counters, and per-phase timing (blocked, connecting, sending, waiting, receiving).

>>> import rampa.http

class rampa.http.Response ¶ class class class rampa.http.Response ¶

Bases: object

Wrapper around an HTTP response with metric-relevant fields.

>>> r = Response(status=200, headers={}, body=b"ok", url="http://x")
>>> r.status
200

rampa.http._estimate_request_size(kwargs) ¶ function function rampa.http._estimate_request_size(kwargs) ¶

Estimate the size of an outgoing request body in bytes.

Parameters:

kwargs (dict[str, Any]) – Request keyword arguments.

Returns:

int – Estimated body size in bytes.
>>> _estimate_request_size({“data” (“hello”}))
5
>>> _estimate_request_size({“data” (b”bytes”}))
5
>>> _estimate_request_size({})
0

Return type:

int

class rampa.http._TraceTimings ¶ class class class rampa.http._TraceTimings ¶

Bases: object

Per-request timing accumulator for aiohttp trace signals.

Accepts and ignores trace_request_ctx so it can be used as an aiohttp trace_config_ctx_factory.

>>> t = _TraceTimings()
>>> t.request_start
0

rampa.http._build_trace_config(client) ¶ function function rampa.http._build_trace_config(client) ¶

Build an aiohttp TraceConfig that records per-phase timings.

The timings are stored on the trace request context object (a _TraceTimings instance) and read back by the caller after the request completes.

>>> import rampa.http

Parameters:: client (Any)
Return type:: TraceConfig

class rampa.http.HttpClient ¶ class class class rampa.http.HttpClient ¶

Bases: object

HTTP client that auto-emits metrics for every request.

Parameters:

sample_queue (queue.SimpleQueue[Sample | None]) – Queue for emitting metric samples.
tags (dict[str, str]) – Base tags added to every sample (e.g. scenario name).
q (>>> import queue as)
sq (>>>)
HttpClient(sq (>>> client =)
{"scenario" ("test"}))
client._tags["scenario"] (>>>)
'test'

Metrics¶

Core data types for the rampa load testing framework.

This module defines the hot-path types used throughout rampa. These types are allocated millions of times during a test run, so they use lightweight Python primitives (NamedTuple, enum) rather than validation-heavy models.

>>> import rampa._types

class rampa._types.MetricType ¶ class class class rampa._types.MetricType ¶

Bases: Enum

Classification of a metric’s aggregation behavior.

Each metric type determines which sink accumulates samples and which aggregations are available for thresholds and summaries.

>>> MetricType.COUNTER.value
'counter'
>>> MetricType.TREND.value
'trend'

class rampa._types.ValueType ¶ class class class rampa._types.ValueType ¶

Bases: Enum

Unit hint for metric display formatting.

>>> ValueType.TIME.value
'time'
>>> ValueType.DEFAULT.value
'default'

class rampa._types.Sample ¶ class class class rampa._types.Sample ¶

Bases: NamedTuple

A single metric observation emitted by a worker or protocol client.

Samples are the hot-path data type. Every HTTP request produces multiple samples (one per timing phase). Construction must be as cheap as possible.

Parameters:

metric (str) – Metric name, e.g. "http_req_duration".
value (float) – Observed value.
timestamp (int) – Monotonic nanoseconds via time.monotonic_ns().
tags (dict[str, str]) – Low-cardinality indexed dimensions.
Sample("http_reqs" (>>> s =)
1.0 ("GET"}))
0 ("GET"}))
{"method" ("GET"}))
s.metric (>>>)
'http_reqs'
s.tags["method"] (>>>)
'GET'

rampa._types.make_sample(metric, value, tags=None) ¶ function function rampa._types.make_sample(metric, value, tags=None) ¶

Create a sample with the current monotonic timestamp.

Parameters:

metric (str) – Metric name.
value (float) – Observed value.
tags (dict[str, str] | None) – Optional tags. Defaults to an empty dict.

Returns:

Sample – A new sample with timestamp set to time.monotonic_ns().
>>> s = make_sample(“iterations”, 1.0)
>>> s.metric
’iterations’
>>> s.value
1.0
>>> isinstance(s.timestamp, int)
True
>>> s.tags
{}

Return type:

Sample

Metric registry, sinks, and aggregation engine for rampa.

The metric engine runs in a dedicated thread, draining samples from a queue.SimpleQueue and updating per-metric sinks. It periodically emits MetricSnapshot objects for outputs and threshold evaluation.

>>> import rampa.metrics

class rampa.metrics.Metric ¶ class class class rampa.metrics.Metric ¶

Bases: object

A registered metric with name, type, and value type.

>>> m = Metric(name="http_reqs", metric_type=MetricType.COUNTER)
>>> m.metric_type
<MetricType.COUNTER: 'counter'>

class rampa.metrics.CounterSink ¶ class class class rampa.metrics.CounterSink ¶

Bases: object

Accumulates a running total and computes rate.

>>> s = CounterSink()
>>> s.add(1.0)
>>> s.add(2.0)
>>> s.format(10.0)
{'count': 3.0, 'rate': 0.3}

class rampa.metrics.GaugeSink ¶ class class class rampa.metrics.GaugeSink ¶

Bases: object

Tracks the latest value plus min and max.

>>> s = GaugeSink()
>>> s.add(10.0)
>>> s.add(5.0)
>>> s.add(20.0)
>>> s.format(1.0)
{'value': 20.0, 'min': 5.0, 'max': 20.0}

class rampa.metrics.RateSink ¶ class class class rampa.metrics.RateSink ¶

Bases: object

Tracks the fraction of non-zero (truthy) values.

>>> s = RateSink()
>>> s.add(1.0)
>>> s.add(0.0)
>>> s.add(1.0)
>>> fmt = s.format(1.0)
>>> fmt["passes"]
2.0
>>> fmt["fails"]
1.0

class rampa.metrics.TrendSink ¶ class class class rampa.metrics.TrendSink ¶

Bases: object

Stores all values for percentile computation.

Uses linear interpolation matching NumPy’s default algorithm.

>>> s = TrendSink()
>>> for v in [10.0, 20.0, 30.0, 40.0, 50.0]:
...     s.add(v)
>>> fmt = s.format(1.0)
>>> fmt["min"]
10.0
>>> fmt["max"]
50.0
>>> fmt["avg"]
30.0
>>> fmt["med"]
30.0
>>> fmt["count"]
5.0

class rampa.metrics.SinkProtocol ¶ class class class rampa.metrics.SinkProtocol ¶

Bases: Protocol

Structural protocol for metric sink implementations.

Enables future Rust PyO3 sinks to satisfy the interface without appearing in a union type.

>>> class DummySink:
...     def add(self, value: float) -> None: ...
...     def format(self, duration: float) -> dict[str, float]:
...         return {}
>>> hasattr(DummySink, "add") and hasattr(DummySink, "format")
True

rampa.metrics.Sink ¶ attribute attribute rampa.metrics.Sink ¶: Type alias for metric sink implementations.

class rampa.metrics.HdrTrendSink ¶ class class class rampa.metrics.HdrTrendSink ¶

Bases: object

Trend sink backed by a Rust HDR histogram.

Fixed ~20KB memory regardless of sample count. O(1) insert, O(1) percentile queries. Requires the rampa._core Rust extension.

Values are stored as integer microseconds internally. The add() method accepts float milliseconds (matching the Python TrendSink interface) and converts automatically.

>>> try:
...     from rampa._core import HdrHistogram
...     s = HdrTrendSink()
...     for v in [10.0, 20.0, 30.0, 40.0, 50.0]:
...         s.add(v)
...     fmt = s.format(1.0)
...     fmt["count"]
... except ImportError:
...     5.0
5.0

rampa.metrics.create_sink(metric_type) ¶ function function rampa.metrics.create_sink(metric_type) ¶

Create a sink matching the metric type.

Uses Rust HDR histogram for trend sinks when available, falling back to the Python TrendSink.

Parameters:

metric_type (MetricType) – The type of metric.

Returns:

Sink – A new sink instance.
>>> type(create_sink(MetricType.COUNTER)).__name__
’CounterSink’
>>> type(create_sink(MetricType.TREND)).__name__ in (‘TrendSink’, ‘HdrTrendSink’)
True

Return type:

Sink

rampa.metrics.SubmetricKey ¶ data data rampa.metrics.SubmetricKey ¶: Identifies a tag-filtered view of a base metric.

The first element is the base metric name, the second is a frozen set of (tag_key, tag_value) pairs used for matching.

alias of tuple[str, frozenset[tuple[str, str]]]

class rampa.metrics.MetricRegistry ¶ class class class rampa.metrics.MetricRegistry ¶

Bases: object

Thread-safe registry that enforces name-type consistency.

>>> reg = MetricRegistry()
>>> reg.get_or_create("reqs", MetricType.COUNTER).metric_type
<MetricType.COUNTER: 'counter'>
>>> reg.get_or_create("reqs", MetricType.COUNTER).name
'reqs'

get_or_create_sub_sink(base_name, tag_filter) ¶ method method get_or_create_sub_sink(base_name, tag_filter) ¶

Get or create a sub-sink for tag-filtered threshold evaluation.

Parameters:

base_name (str) – Base metric name.
tag_filter (dict[str, str]) – Tags that samples must match to feed this sub-sink.

Returns:

Sink | None – The sub-sink, or None if the base metric is not registered.
>>> reg = MetricRegistry()
>>> _ = reg.get_or_create(“dur”, MetricType.TREND)
>>> sink = reg.get_or_create_sub_sink(“dur”, {“status” (“200”}))
>>> type(sink).__name__ in (‘TrendSink’, ‘HdrTrendSink’)
True

Return type:

Sink | None

class rampa.metrics.MetricSnapshot ¶ class class class rampa.metrics.MetricSnapshot ¶

Bases: object

Frozen snapshot of aggregated metric values at a point in time.

>>> snap = MetricSnapshot(
...     timestamp=0,
...     duration=10.0,
...     values={"reqs": {"count": 100.0, "rate": 10.0}},
... )
>>> snap.values["reqs"]["rate"]
10.0

rampa.metrics.register_builtins(registry) ¶ function function rampa.metrics.register_builtins(registry) ¶

Parameters:

registry (MetricRegistry) – The registry to populate.
MetricRegistry() (>>> reg =)
register_builtins(reg) (>>>)
None (>>> reg.get_sink("checks") is not)
True
None
True

Return type:

None

class rampa.metrics.MetricEngine ¶ class class class rampa.metrics.MetricEngine ¶

Bases: object

Background thread that drains samples and updates sinks.

Parameters:

registry (MetricRegistry) – Metric registry with sinks.
sample_queue (queue.SimpleQueue[Sample | None]) – Queue of samples. Send None to signal shutdown.
flush_interval (float) – Seconds between snapshot emissions.

Examples

>>> eng = MetricEngine(
...     registry=MetricRegistry(),
...     sample_queue=queue.SimpleQueue(),
... )
>>> eng.flush_interval
0.05

__init__(registry, sample_queue, flush_interval=0.05, on_sample=None, on_snapshot=None, thresholds={}, threshold_interval=2.0, on_threshold=None, abort_callback=None) ¶ method method __init__(registry, sample_queue, flush_interval=0.05, on_sample=None, on_snapshot=None, thresholds={}, threshold_interval=2.0, on_threshold=None, abort_callback=None) ¶

Parameters:

registry (MetricRegistry)
sample_queue (SimpleQueue[Sample | None])
flush_interval (float)
on_sample (Callable[[Sample], None] | None)
on_snapshot (Callable[[MetricSnapshot], None] | None)
thresholds (dict[str, list[Any]])
threshold_interval (float)
on_threshold (Callable[[list[Any]], None] | None)
abort_callback (Callable[[], None] | None)

Return type:

None

Outputs¶

class rampa.output.Output ¶ class class class rampa.output.Output ¶

Bases: Protocol

Protocol for metric output backends.

Outputs receive batched samples on a periodic flush interval. They must not block the engine’s sample fan-out path.

class rampa.output.OutputManager ¶ class class class rampa.output.OutputManager ¶

Bases: object

Fans out sample batches to all registered outputs.

>>> import asyncio
>>> mgr = OutputManager()
>>> len(mgr.outputs)
0

rampa.outputs.get_output(name, destination='') ¶ function function rampa.outputs.get_output(name, destination='') ¶

Resolve an output backend by name.

Parameters:

name (str) – Backend name (e.g. "json", "csv").
destination (str) – Backend-specific destination (file path, URL, etc.).

Returns:

An output instance.

Return type:

Output

Raises:

ValueError – If the backend name is not registered.
>>> type(get_output("json", "/dev/null")).__name__ –
'JSONOutput' –

Protocol Clients¶

WebSocket protocol client with automatic metric emission.

Emits ws_sessions, ws_connecting, ws_session_duration, ws_messages_sent, ws_messages_received, and ws_errors metrics.

>>> import rampa.protocols.websocket

class rampa.protocols.websocket.WebSocketSession ¶ class class class rampa.protocols.websocket.WebSocketSession ¶

Bases: object

An active WebSocket connection with metric emission.

Created by WebSocketClient.connect(). Use as an async context manager for automatic cleanup.

>>> import rampa.protocols.websocket

class rampa.protocols.websocket.WebSocketClient ¶ class class class rampa.protocols.websocket.WebSocketClient ¶

Bases: object

WebSocket client with automatic metric emission.

Lazily initialized via worker.ws. Uses aiohttp’s WebSocket client under the hood.

>>> import rampa.protocols.websocket

gRPC protocol client with automatic metric emission.

Emits grpc_reqs, grpc_req_duration, grpc_req_failed, grpc_streams_opened, grpc_messages_sent, and grpc_messages_received metrics.

Requires the grpcio optional dependency.

>>> import rampa.protocols.grpc

class rampa.protocols.grpc.GrpcResponse ¶ class class class rampa.protocols.grpc.GrpcResponse ¶

Bases: object

Response from a gRPC unary call.

>>> r = GrpcResponse(data=b"ok", status_code=0, metadata={})
>>> r.status_code
0

class rampa.protocols.grpc.GrpcClient ¶ class class class rampa.protocols.grpc.GrpcClient ¶

Bases: object

gRPC client with automatic metric emission.

Uses grpc.aio for async RPC calls. Lazily initialized via worker.grpc.

>>> import rampa.protocols.grpc

Thresholds¶

Threshold expression parser and evaluator for rampa.

Thresholds define pass/fail criteria over metric aggregates. They are the policy layer — checks measure facts, thresholds decide final status.

>>> import rampa.thresholds

class rampa.thresholds.ThresholdExpression ¶ class class class rampa.thresholds.ThresholdExpression ¶

Bases: object

A parsed threshold expression.

>>> expr = parse_threshold("p(95)<500")
>>> expr.aggregation
'p(95)'
>>> expr.operator
'<'
>>> expr.value
500.0

rampa.thresholds.parse_threshold(expr) ¶ function function rampa.thresholds.parse_threshold(expr) ¶

Parse a threshold expression string.

Parameters:

expr (str) – Expression like "p(95)<500", "rate<0.01", "avg>200".

Returns:

Parsed expression.

Return type:

ThresholdExpression

Raises:

ValueError – If the expression is malformed.
>>> parse_threshold("count>=100").aggregation –
'count' –
>>> parse_threshold("med<400").value –
400.0 –
>>> parse_threshold("p(99.9)<=1000").aggregation –
'p(99.9)' –
>>> parse_threshold("p(99.9)<=1000").aggregation_value –
99.9 –

rampa.thresholds.parse_submetric(name) ¶ function function rampa.thresholds.parse_submetric(name) ¶

Parse a metric name that may include tag filters.

Parameters:

name (str) – Metric name, optionally with tag filters like "http_req_duration{status:200}".

Returns:

tuple[str, dict[str, str]] – The base metric name and a dict of tag filters.
>>> parse_submetric(“http_req_duration{status (200}”))
(‘http_req_duration’, {‘status’ (‘200’}))
>>> parse_submetric(“http_reqs”)
(‘http_reqs’, {})
>>> parse_submetric(“dur{method (GET,name:login}”))
(‘dur’, {‘method’ (‘GET’, ‘name’: ‘login’}))

Return type:

tuple[str, dict[str, str]]

class rampa.thresholds.Threshold ¶ class class class rampa.thresholds.Threshold ¶

Bases: object

A configured threshold with expression and abort behavior.

>>> t = Threshold(
...     source="p(95)<500",
...     expression=parse_threshold("p(95)<500"),
... )
>>> t.abort_on_fail
False

class rampa.thresholds.ThresholdResult ¶ class class class rampa.thresholds.ThresholdResult ¶

Bases: object

Result of evaluating one threshold.

>>> r = ThresholdResult(source="p(95)<500", passed=True, lhs=450.0, rhs=500.0)
>>> r.passed
True

rampa.thresholds._sink_key(aggregation) ¶ function function rampa.thresholds._sink_key(aggregation) ¶

Map threshold aggregation to sink format key.

Parameters:: aggregation (str)
Return type:: str

rampa.thresholds.evaluate_threshold(threshold, sink, duration) ¶ function function rampa.thresholds.evaluate_threshold(threshold, sink, duration) ¶

Evaluate a single threshold against a sink.

Parameters:

threshold (Threshold) – The threshold to evaluate.
sink (Sink) – The metric sink containing aggregated values.
duration (float) – Elapsed test duration in seconds.

Returns:

ThresholdResult – Whether the threshold passed.
>>> from rampa.metrics import TrendSink
>>> s = TrendSink()
>>> for v in range(100)
… s.add(float(v))
>>> t = Threshold(
… source=”p(95)<200”,
… expression=parse_threshold(“p(95)<200”),
… )
>>> evaluate_threshold(t, s, 1.0).passed
True

Return type:

ThresholdResult

rampa.thresholds.evaluate_thresholds(metric_thresholds, sinks, duration, sub_sinks=None) ¶ function function rampa.thresholds.evaluate_thresholds(metric_thresholds, sinks, duration, sub_sinks=None) ¶

Evaluate all thresholds against their metric sinks.

Parameters:

metric_thresholds (dict[str, list[Threshold]]) – Metric name → list of thresholds for that metric. Names may include a tag filter suffix like "dur{status:200}".
sinks (dict[str, Sink]) – Base metric name → sink mapping.
duration (float) – Elapsed test duration in seconds.
sub_sinks (dict[..., Sink] | None) – Optional submetric (base_name, frozenset tags) → sink mapping for tag-filtered threshold evaluation.

Returns:

list[ThresholdResult] – Results for each threshold.
>>> from rampa.metrics import CounterSink
>>> sink = CounterSink()
>>> sink.add(50.0)
>>> mt = {“reqs” ([Threshold()
… source=”count>=100”,
… expression=parse_threshold(“count>=100”),
… )]}
>>> results = evaluate_thresholds(mt, {“reqs” (sink}, 1.0))
>>> results[0].passed
False

Return type:

list[ThresholdResult]

Distributed Execution¶

Execution segments for deterministic work partitioning.

Segments divide VUs, iterations, and arrival rates across workers without central assignment. Each worker independently computes its share from its index and the total worker count.

>>> import rampa.distributed.segment

class rampa.distributed.segment.ExecutionSegment ¶ class class class rampa.distributed.segment.ExecutionSegment ¶

Bases: object

Integer-range partition for deterministic VU/iteration assignment.

For N workers running K total VUs, worker i gets VUs [i*K//N, (i+1)*K//N). The same logic applies to iterations and arrival rates.

Parameters:

index (int) – Zero-based worker index.
total (int) – Total number of workers in the cluster.

Examples

>>> seg = ExecutionSegment(index=0, total=3)
>>> seg.vu_range(10)
range(0, 3)
>>> seg = ExecutionSegment(index=1, total=3)
>>> seg.vu_range(10)
range(3, 6)
>>> seg = ExecutionSegment(index=2, total=3)
>>> seg.vu_range(10)
range(6, 10)

scale_rate(total_rate, total_vus) ¶ method method scale_rate(total_rate, total_vus) ¶

Return this segment’s share of a total arrival rate.

Parameters:

total_rate (float) – Desired total arrival rate (e.g. requests/sec).
total_vus (int) – Total virtual users being partitioned.

Returns:

float – This segment’s proportional rate.
>>> ExecutionSegment(0, 4).scale_rate(100.0, 4)
25.0
>>> ExecutionSegment(0, 3).scale_rate(100.0, 3) # doctest (+ELLIPSIS)
33.3…
>>> ExecutionSegment(0, 2).scale_rate(100.0, 5)
40.0
>>> ExecutionSegment(1, 2).scale_rate(100.0, 5)
60.0

Return type:

float

Self-contained test archives for distributed execution.

A .rampa file is a zip archive containing the test script, data files, dependency list, and a manifest for reproducibility.

>>> import rampa.distributed.archive

class rampa.distributed.archive.ArchiveManifest ¶ class class class rampa.distributed.archive.ArchiveManifest ¶

Bases: object

Metadata for a rampa test archive.

>>> m = ArchiveManifest(entrypoint="load_test.py")
>>> m.version
1

rampa.distributed.archive.create_archive(script_path, output_path, data_files=None, requirements=None) ¶ function function rampa.distributed.archive.create_archive(script_path, output_path, data_files=None, requirements=None) ¶

Create a self-contained .rampa test archive.

Parameters:

script_path (str | Path) – Path to the test script.
output_path (str | Path) – Output .rampa file path.
data_files (list[str | Path] | None) – Additional data files to include.
requirements (list[str] | None) – pip-compatible dependency strings.

Returns:

Path – The created archive path.
>>> import tempfile, pathlib
>>> with tempfile.TemporaryDirectory() as d
… script = pathlib.Path(d) / “test.py”
… _ = script.write_text(“async def default(w) (pass”))
… out = pathlib.Path(d) / “test.rampa”
… result = create_archive(script, out)
… result.suffix
’.rampa’

Return type:

Path

rampa.distributed.archive.extract_archive(archive_path, target_dir) ¶ function function rampa.distributed.archive.extract_archive(archive_path, target_dir) ¶

Extract a .rampa archive and return its manifest.

Parameters:

archive_path (str | Path) – Path to the .rampa file.
target_dir (str | Path) – Directory to extract into.

Returns:

ArchiveManifest – Parsed manifest from the archive.
>>> import tempfile, pathlib
>>> with tempfile.TemporaryDirectory() as d
… script = pathlib.Path(d) / “test.py”
… _ = script.write_text(“async def default(w) (pass”))
… archive = pathlib.Path(d) / “test.rampa”
… _ = create_archive(script, archive)
… out_dir = pathlib.Path(d) / “extracted”
… manifest = extract_archive(archive, out_dir)
… manifest.entrypoint
’script.py’

Return type:

ArchiveManifest

CI Comparison¶

Compare two rampa JSON result files for CI benchmarking.

Reads summary metrics from JSON output files and produces comparison tables in text, markdown, or JSON format.

>>> import rampa.ci.compare

class rampa.ci.compare.MetricDelta ¶ class class class rampa.ci.compare.MetricDelta ¶

Bases: NamedTuple

Comparison of a single metric stat between two runs.

>>> d = MetricDelta("http_req_duration", "p(95)", 45.0, 52.0, 15.56)
>>> d.pct_change
15.56

rampa.ci.compare.is_regressed(delta) ¶ function function rampa.ci.compare.is_regressed(delta) ¶

Return True if the metric got worse (>5% increase).

>>> is_regressed(MetricDelta("dur", "p95", 45.0, 52.0, 15.56))
True
>>> is_regressed(MetricDelta("dur", "p95", 45.0, 46.0, 2.2))
False

Parameters:: delta (MetricDelta)
Return type:: bool

rampa.ci.compare.compare_results(baseline_path, current_path, metrics=None) ¶ function function rampa.ci.compare.compare_results(baseline_path, current_path, metrics=None) ¶

Compare metrics between two JSON result files.

Parameters:

baseline_path (str | Path) – Path to the baseline JSON result.
current_path (str | Path) – Path to the current JSON result.
metrics (list[str] | None) – Metric names to compare. None compares all shared metrics.

Returns:

list[MetricDelta] – Per-stat deltas for each metric.
>>> import tempfile, pathlib, json
>>> with tempfile.TemporaryDirectory() as d
… base = pathlib.Path(d) / “base.json”
… curr = pathlib.Path(d) / “curr.json”
… data = {“summary” ({“metrics”: {“reqs”: {“count”: 100.0}}}})
… _ = base.write_text(json.dumps(data))
… data[“summary”][“metrics”][“reqs”][“count”] = 120.0
… _ = curr.write_text(json.dumps(data))
… deltas = compare_results(base, curr)
… deltas[0].pct_change
20.0

Return type:

list[MetricDelta]

rampa.ci.compare.format_text(deltas) ¶ function function rampa.ci.compare.format_text(deltas) ¶

Format deltas as a plain text table.

>>> deltas = [MetricDelta("dur", "p95", 45.0, 52.0, 15.56)]
>>> "dur" in format_text(deltas)
True

Parameters:: deltas (list[MetricDelta])
Return type:: str

rampa.ci.compare.format_markdown(deltas) ¶ function function rampa.ci.compare.format_markdown(deltas) ¶

Format deltas as a GitHub-flavored markdown table.

>>> deltas = [MetricDelta("dur", "p95", 45.0, 52.0, 15.56)]
>>> "| dur" in format_markdown(deltas)
True

Parameters:: deltas (list[MetricDelta])
Return type:: str

rampa.ci.compare.format_json(deltas) ¶ function function rampa.ci.compare.format_json(deltas) ¶

Format deltas as JSON.

>>> deltas = [MetricDelta("dur", "p95", 45.0, 52.0, 15.56)]
>>> import json
>>> data = json.loads(format_json(deltas))
>>> data[0]["metric"]
'dur'

Parameters:: deltas (list[MetricDelta])
Return type:: str

Loader¶

Test script loader and @scenario decorator for rampa.

Discovers scenario functions, setup, and teardown from user Python modules.

>>> import rampa.loader

rampa.loader.scenario(name=None, **kwargs)
¶
function
function
rampa.loader.scenario(name=None, **kwargs)
¶

Mark an async function as a rampa scenario.

Parameters:

name (str | None) – Scenario name. Defaults to the function name.
**kwargs (Any) – ScenarioConfig fields (executor, vus, duration, etc.).

Returns:

Callable – The decorated function with scenario metadata attached.
>>> @scenario(executor=”constant-vus”, vus=5, duration=”10s”)
… async def load_test(worker (object) -> None:)
… pass
>>> hasattr(load_test, “_rampa_scenario_config”)
True

Return type:

Callable[[Callable[..., Any]], Callable[..., Any]]

class rampa.loader.TestPlan ¶ class class class rampa.loader.TestPlan ¶

Bases: object

Resolved test plan ready for execution.

>>> plan = TestPlan(scenarios={}, config=Config())
>>> plan.setup_fn is None
True

rampa.loader.load_test(path) ¶ function function rampa.loader.load_test(path) ¶

Load a test module and extract the test plan.

Parameters:

path (str) – Path to the Python test script.

Returns:

Resolved test plan with scenarios, setup, teardown, and config.

Return type:

TestPlan

Raises:

FileNotFoundError – If the script file does not exist.
ValueError – If no scenarios are found.