benchmatrix¶

benchmatrix ¶

Public pytest-benchmark matrix and JSON results API.

TargetFunction `module-attribute` ¶

TargetFunction: TypeAlias = Callable[..., object]

Synchronous callable measured by pytest-benchmark through benchmatrix.

Target functions must perform the work being measured before returning. Async functions are rejected. Lazy return values are not forced by the harness.

BenchmarkCase `dataclass` ¶

Named input case and metadata for a pytest-benchmark matrix.

Warning

If fresh_inputs is false, pytest-benchmark may call the target function repeatedly with the same argument objects. That is appropriate only when the target function treats its inputs as immutable or when reuse reflects the workload you want to measure.

If fresh_inputs is true, this harness uses pytest-benchmark pedantic setup so input construction is setup work rather than timed target-function work. That avoids accidentally timing input creation, but it also means the benchmark is not an end-to-end measurement that includes input construction. To benchmark construction cost, put that construction inside the target function itself.

When fresh_inputs is true, BenchmarkConfig.pedantic_iterations is ignored because pytest-benchmark setup mode is used. The harness emits a runtime warning when a non-default value is ignored.

Parameters:

Name	Type	Description	Default
`name`	`str`	Human-readable case name used in parameter IDs and metadata.	required
`make_args`	`Callable[[], tuple[object, ...]]`	Factory returning positional arguments for the target function.	`_empty_args`
`make_kwargs`	`Callable[[], dict[str, object]]`	Factory returning keyword arguments for the target function.	`_empty_kwargs`
`work_units`	`float \| Callable[[], float] \| None`	Positive logical amount of work performed by one target call. This can represent items, rows, bytes, tokens, records, events, or any other domain-specific unit.	`None`
`work_unit_name`	`str`	Name of the logical work unit, such as `"items"`, `"rows"`, `"bytes"`, or `"tokens"`. Use a base unit name without spaces, slashes, or `"/s"`; display code appends `"/s"` for throughput.	`_DEFAULT_WORK_UNIT_NAME`
`fresh_inputs`	`bool`	Whether each benchmark round needs newly created inputs.	`False`
`metadata`	`Mapping[str, object]`	Additional strict-JSON-renderable metadata describing the case. Reasonable scalar types such as paths, datetimes, enums, and NumPy scalars are coerced; unsupported values raise `MetadataSerializationError`.	`_empty_metadata()`

Attributes:

Name	Type	Description
`name`	`str`	Human-readable case name used in parameter IDs and metadata.
`make_args`	`Callable[[], tuple[object, ...]]`	Factory returning positional arguments for the target function.
`make_kwargs`	`Callable[[], dict[str, object]]`	Factory returning keyword arguments for the target function.
`work_units`	`float \| Callable[[], float] \| None`	Positive logical amount of work performed by one target call.
`work_unit_name`	`str`	Name of the logical work unit.
`fresh_inputs`	`bool`	Whether each benchmark round needs newly created inputs.
`metadata`	`Mapping[str, object]`	Strict JSON-safe metadata describing the case.

Source code in src/benchmatrix/bench_harness.py

@dataclass(frozen=True, slots=True)
class BenchmarkCase:
    """Named input case and metadata for a pytest-benchmark matrix.

    Warning:
        If ``fresh_inputs`` is false, pytest-benchmark may call the target
        function repeatedly with the same argument objects. That is appropriate
        only when the target function treats its inputs as immutable or when
        reuse reflects the workload you want to measure.

        If ``fresh_inputs`` is true, this harness uses pytest-benchmark
        pedantic setup so input construction is setup work rather than timed
        target-function work. That avoids accidentally timing input creation,
        but it also means the benchmark is not an end-to-end measurement that
        includes input construction. To benchmark construction cost, put that
        construction inside the target function itself.

        When ``fresh_inputs`` is true, ``BenchmarkConfig.pedantic_iterations``
        is ignored because pytest-benchmark setup mode is used. The harness
        emits a runtime warning when a non-default value is ignored.

    Args:
        name: Human-readable case name used in parameter IDs and metadata.
        make_args: Factory returning positional arguments for the target
            function.
        make_kwargs: Factory returning keyword arguments for the target
            function.
        work_units: Positive logical amount of work performed by one target
            call. This can represent items, rows, bytes, tokens, records,
            events, or any other domain-specific unit.
        work_unit_name: Name of the logical work unit, such as ``"items"``,
            ``"rows"``, ``"bytes"``, or ``"tokens"``. Use a base unit name
            without spaces, slashes, or ``"/s"``; display code appends ``"/s"``
            for throughput.
        fresh_inputs: Whether each benchmark round needs newly created inputs.
        metadata: Additional strict-JSON-renderable metadata describing the
            case. Reasonable scalar types such as paths, datetimes, enums, and
            NumPy scalars are coerced; unsupported values raise
            ``MetadataSerializationError``.

    Attributes:
        name: Human-readable case name used in parameter IDs and metadata.
        make_args: Factory returning positional arguments for the target
            function.
        make_kwargs: Factory returning keyword arguments for the target
            function.
        work_units: Positive logical amount of work performed by one target
            call.
        work_unit_name: Name of the logical work unit.
        fresh_inputs: Whether each benchmark round needs newly created inputs.
        metadata: Strict JSON-safe metadata describing the case.
    """

    name: str
    make_args: Callable[[], tuple[object, ...]] = _empty_args
    make_kwargs: Callable[[], dict[str, object]] = _empty_kwargs
    work_units: float | Callable[[], float] | None = None
    work_unit_name: str = _DEFAULT_WORK_UNIT_NAME
    fresh_inputs: bool = False
    metadata: Mapping[str, object] = field(default_factory=_empty_metadata)

    def __post_init__(self) -> None:
        """Validate benchmark case fields after initialization."""
        object.__setattr__(self, "name", _validate_name(self.name, field="case name"))

        object.__setattr__(self, "work_unit_name", _validate_work_unit_name(self.work_unit_name))

        if self.work_units is not None and not callable(self.work_units):
            _ = _validate_work_units(self.work_units)

        coerced_metadata = _coerce_json_mapping(
            self.metadata,
            path="BenchmarkCase.metadata",
        )
        object.__setattr__(self, "metadata", coerced_metadata)

    def make_call(self) -> tuple[tuple[object, ...], dict[str, object]]:
        """Return positional and keyword arguments for one target invocation.

        Returns:
            A tuple containing positional arguments and keyword arguments.
        """
        return self.make_args(), self.make_kwargs()

    def work_unit_count(self) -> float | None:
        """Return the logical work-unit count for throughput metrics.

        Returns:
            The logical work-unit count, or ``None`` when the case has no work
            unit count.

        Raises:
            ValueError: If the work-unit count is not positive or finite.
        """
        if self.work_units is None:
            return None

        value = self.work_units() if callable(self.work_units) else self.work_units
        return _validate_work_units(value)

    @classmethod
    def from_values(
        cls,
        name: str,
        *args: object,
        work_units: float | Callable[[], float] | None = None,
        work_unit_name: str = _DEFAULT_WORK_UNIT_NAME,
        fresh_inputs: bool = False,
        copier: Callable[[object], object] | None = None,
        metadata: Mapping[str, object] | None = None,
        **kwargs: object,
    ) -> BenchmarkCase:
        """Create a benchmark case from concrete argument values.

        Args:
            name: Case name.
            *args: Positional arguments for the target function.
            work_units: Positive logical amount of work performed by one target
                call.
            work_unit_name: Name of the logical work unit, such as ``"items"``,
                ``"rows"``, ``"bytes"``, or ``"tokens"``. Use a base unit name
                without spaces, slashes, or ``"/s"``.
            fresh_inputs: Whether target invocations need fresh inputs. When
                true and ``copier`` is omitted, a shallow copy is made for each
                argument value.
            copier: Optional copy function applied to each argument value. Use
                ``deep_copy`` or a domain-specific copy function when shallow
                copies are not fresh enough for the benchmarked workload.
            metadata: Optional strict-JSON-renderable case metadata.
            **kwargs: Keyword arguments for the target function.

        Returns:
            A configured benchmark case.
        """

        effective_copier = shallow_copy if fresh_inputs and copier is None else copier

        def make_args() -> tuple[object, ...]:
            """Return case positional arguments."""
            if effective_copier is None:
                return args

            return tuple(effective_copier(arg) for arg in args)

        def make_kwargs() -> dict[str, object]:
            """Return case keyword arguments."""
            if effective_copier is None:
                return dict(kwargs)

            return {key: effective_copier(value) for key, value in kwargs.items()}

        return cls(
            name=name,
            make_args=make_args,
            make_kwargs=make_kwargs,
            work_units=work_units,
            work_unit_name=work_unit_name,
            fresh_inputs=fresh_inputs or effective_copier is not None,
            metadata={} if metadata is None else dict(metadata),
        )

__post_init__ ¶

__post_init__() -> None

Validate benchmark case fields after initialization.

Source code in src/benchmatrix/bench_harness.py

def __post_init__(self) -> None:
    """Validate benchmark case fields after initialization."""
    object.__setattr__(self, "name", _validate_name(self.name, field="case name"))

    object.__setattr__(self, "work_unit_name", _validate_work_unit_name(self.work_unit_name))

    if self.work_units is not None and not callable(self.work_units):
        _ = _validate_work_units(self.work_units)

    coerced_metadata = _coerce_json_mapping(
        self.metadata,
        path="BenchmarkCase.metadata",
    )
    object.__setattr__(self, "metadata", coerced_metadata)

make_call ¶

make_call() -> tuple[tuple[object, ...], dict[str, object]]

Return positional and keyword arguments for one target invocation.

Returns:

Type	Description
`tuple[tuple[object, ...], dict[str, object]]`	A tuple containing positional arguments and keyword arguments.

Source code in src/benchmatrix/bench_harness.py

def make_call(self) -> tuple[tuple[object, ...], dict[str, object]]:
    """Return positional and keyword arguments for one target invocation.

    Returns:
        A tuple containing positional arguments and keyword arguments.
    """
    return self.make_args(), self.make_kwargs()

work_unit_count ¶

work_unit_count() -> float | None

Return the logical work-unit count for throughput metrics.

Returns:

Type	Description
`float \| None`	The logical work-unit count, or `None` when the case has no work
`float \| None`	unit count.

Raises:

Type	Description
`ValueError`	If the work-unit count is not positive or finite.

Source code in src/benchmatrix/bench_harness.py

def work_unit_count(self) -> float | None:
    """Return the logical work-unit count for throughput metrics.

    Returns:
        The logical work-unit count, or ``None`` when the case has no work
        unit count.

    Raises:
        ValueError: If the work-unit count is not positive or finite.
    """
    if self.work_units is None:
        return None

    value = self.work_units() if callable(self.work_units) else self.work_units
    return _validate_work_units(value)

from_values `classmethod` ¶

from_values(
    name: str,
    *args: object,
    work_units: float | Callable[[], float] | None = None,
    work_unit_name: str = _DEFAULT_WORK_UNIT_NAME,
    fresh_inputs: bool = False,
    copier: Callable[[object], object] | None = None,
    metadata: Mapping[str, object] | None = None,
    **kwargs: object,
) -> BenchmarkCase

Create a benchmark case from concrete argument values.

Parameters:

Name	Type	Description	Default
`name`	`str`	Case name.	required
`*args`	`object`	Positional arguments for the target function.	`()`
`work_units`	`float \| Callable[[], float] \| None`	Positive logical amount of work performed by one target call.	`None`
`work_unit_name`	`str`	Name of the logical work unit, such as `"items"`, `"rows"`, `"bytes"`, or `"tokens"`. Use a base unit name without spaces, slashes, or `"/s"`.	`_DEFAULT_WORK_UNIT_NAME`
`fresh_inputs`	`bool`	Whether target invocations need fresh inputs. When true and `copier` is omitted, a shallow copy is made for each argument value.	`False`
`copier`	`Callable[[object], object] \| None`	Optional copy function applied to each argument value. Use `deep_copy` or a domain-specific copy function when shallow copies are not fresh enough for the benchmarked workload.	`None`
`metadata`	`Mapping[str, object] \| None`	Optional strict-JSON-renderable case metadata.	`None`
`**kwargs`	`object`	Keyword arguments for the target function.	`{}`

Returns:

Type	Description
`BenchmarkCase`	A configured benchmark case.

Source code in src/benchmatrix/bench_harness.py

@classmethod
def from_values(
    cls,
    name: str,
    *args: object,
    work_units: float | Callable[[], float] | None = None,
    work_unit_name: str = _DEFAULT_WORK_UNIT_NAME,
    fresh_inputs: bool = False,
    copier: Callable[[object], object] | None = None,
    metadata: Mapping[str, object] | None = None,
    **kwargs: object,
) -> BenchmarkCase:
    """Create a benchmark case from concrete argument values.

    Args:
        name: Case name.
        *args: Positional arguments for the target function.
        work_units: Positive logical amount of work performed by one target
            call.
        work_unit_name: Name of the logical work unit, such as ``"items"``,
            ``"rows"``, ``"bytes"``, or ``"tokens"``. Use a base unit name
            without spaces, slashes, or ``"/s"``.
        fresh_inputs: Whether target invocations need fresh inputs. When
            true and ``copier`` is omitted, a shallow copy is made for each
            argument value.
        copier: Optional copy function applied to each argument value. Use
            ``deep_copy`` or a domain-specific copy function when shallow
            copies are not fresh enough for the benchmarked workload.
        metadata: Optional strict-JSON-renderable case metadata.
        **kwargs: Keyword arguments for the target function.

    Returns:
        A configured benchmark case.
    """

    effective_copier = shallow_copy if fresh_inputs and copier is None else copier

    def make_args() -> tuple[object, ...]:
        """Return case positional arguments."""
        if effective_copier is None:
            return args

        return tuple(effective_copier(arg) for arg in args)

    def make_kwargs() -> dict[str, object]:
        """Return case keyword arguments."""
        if effective_copier is None:
            return dict(kwargs)

        return {key: effective_copier(value) for key, value in kwargs.items()}

    return cls(
        name=name,
        make_args=make_args,
        make_kwargs=make_kwargs,
        work_units=work_units,
        work_unit_name=work_unit_name,
        fresh_inputs=fresh_inputs or effective_copier is not None,
        metadata={} if metadata is None else dict(metadata),
    )

BenchmarkConfig `dataclass` ¶

Configuration passed from benchmatrix to pytest-benchmark.

Parameters:

Name	Type	Description	Default
`pedantic_rounds`	`int`	Number of pedantic benchmark rounds to request.	`_DEFAULT_PEDANTIC_ROUNDS`
`warmup_rounds`	`int`	Number of pedantic warmup rounds to request.	`_DEFAULT_WARMUP_ROUNDS`
`pedantic_iterations`	`int`	Number of function calls per pedantic round when inputs are reused. This value is intentionally ignored when `BenchmarkCase.fresh_inputs` is true because pytest-benchmark setup mode is used to keep input construction outside the timed target-function body.	`_DEFAULT_PEDANTIC_ITERATIONS`
`stream_progress`	`bool`	Whether benchmark helpers should print one progress line per benchmark invocation.	`True`

Attributes:

Name	Type	Description
`pedantic_rounds`	`int`	Number of pedantic benchmark rounds to request.
`warmup_rounds`	`int`	Number of pedantic warmup rounds to request.
`pedantic_iterations`	`int`	Number of function calls per pedantic round when inputs are reused.
`stream_progress`	`bool`	Whether benchmark helpers should print one progress line per benchmark invocation.

Raises:

Type	Description
`ValueError`	If rounds or iterations are not positive, or if warmup rounds are negative.

Warning

For tail_latency benchmarks, setting pedantic_iterations above one means raw samples should be interpreted as per-round aggregate timings rather than clean one-call latency samples. The harness emits a runtime warning for this configuration.

Source code in src/benchmatrix/bench_harness.py

@dataclass(frozen=True, slots=True)
class BenchmarkConfig:
    """Configuration passed from benchmatrix to pytest-benchmark.

    Args:
        pedantic_rounds: Number of pedantic benchmark rounds to request.
        warmup_rounds: Number of pedantic warmup rounds to request.
        pedantic_iterations: Number of function calls per pedantic round when
            inputs are reused. This value is intentionally ignored when
            ``BenchmarkCase.fresh_inputs`` is true because pytest-benchmark
            setup mode is used to keep input construction outside the timed
            target-function body.
        stream_progress: Whether benchmark helpers should print one progress
            line per benchmark invocation.

    Attributes:
        pedantic_rounds: Number of pedantic benchmark rounds to request.
        warmup_rounds: Number of pedantic warmup rounds to request.
        pedantic_iterations: Number of function calls per pedantic round when
            inputs are reused.
        stream_progress: Whether benchmark helpers should print one progress
            line per benchmark invocation.

    Raises:
        ValueError: If rounds or iterations are not positive, or if warmup
            rounds are negative.

    Warning:
        For ``tail_latency`` benchmarks, setting ``pedantic_iterations`` above
        one means raw samples should be interpreted as per-round aggregate
        timings rather than clean one-call latency samples. The harness emits a
        runtime warning for this configuration.
    """

    pedantic_rounds: int = _DEFAULT_PEDANTIC_ROUNDS
    warmup_rounds: int = _DEFAULT_WARMUP_ROUNDS
    pedantic_iterations: int = _DEFAULT_PEDANTIC_ITERATIONS
    stream_progress: bool = True

    def __post_init__(self) -> None:
        """Validate benchmark configuration after initialization."""
        if self.pedantic_rounds <= 0:
            raise ValueError("BenchmarkConfig.pedantic_rounds must be positive.")

        if self.warmup_rounds < 0:
            raise ValueError("BenchmarkConfig.warmup_rounds must be non-negative.")

        if self.pedantic_iterations <= 0:
            raise ValueError("BenchmarkConfig.pedantic_iterations must be positive.")

__post_init__ ¶

__post_init__() -> None

Validate benchmark configuration after initialization.

Source code in src/benchmatrix/bench_harness.py

def __post_init__(self) -> None:
    """Validate benchmark configuration after initialization."""
    if self.pedantic_rounds <= 0:
        raise ValueError("BenchmarkConfig.pedantic_rounds must be positive.")

    if self.warmup_rounds < 0:
        raise ValueError("BenchmarkConfig.warmup_rounds must be non-negative.")

    if self.pedantic_iterations <= 0:
        raise ValueError("BenchmarkConfig.pedantic_iterations must be positive.")

BenchmarkFixture ¶

Bases: Protocol

pytest-benchmark fixture surface used by benchmatrix.

Attributes:

Name	Type	Description
`extra_info`	`MutableMapping[str, object]`	Mutable metadata attached to pytest-benchmark output.

Source code in src/benchmatrix/bench_harness.py

class BenchmarkFixture(Protocol):
    """pytest-benchmark fixture surface used by benchmatrix.

    Attributes:
        extra_info: Mutable metadata attached to pytest-benchmark output.
    """

    extra_info: MutableMapping[str, object]

    def __call__(self, target: Callable[..., T], *args: object, **kwargs: object) -> T:
        """Benchmark ``target`` with pytest-benchmark automatic calibration."""
        ...

    def pedantic(
        self,
        target: Callable[..., T],
        *,
        args: Sequence[object] | None = None,
        kwargs: Mapping[str, object] | None = None,
        setup: Callable[[], tuple[Sequence[object], Mapping[str, object]]] | None = None,
        teardown: Callable[..., object] | None = None,
        rounds: int = _DEFAULT_PEDANTIC_ROUNDS,
        warmup_rounds: int = _DEFAULT_WARMUP_ROUNDS,
        iterations: int = _DEFAULT_PEDANTIC_ITERATIONS,
    ) -> T:
        """Benchmark ``target`` with pytest-benchmark pedantic mode."""
        ...

call ¶

__call__(
    target: Callable[..., T],
    *args: object,
    **kwargs: object,
) -> T

Benchmark target with pytest-benchmark automatic calibration.

Source code in src/benchmatrix/bench_harness.py

def __call__(self, target: Callable[..., T], *args: object, **kwargs: object) -> T:
    """Benchmark ``target`` with pytest-benchmark automatic calibration."""
    ...

pedantic ¶

pedantic(
    target: Callable[..., T],
    *,
    args: Sequence[object] | None = None,
    kwargs: Mapping[str, object] | None = None,
    setup: Callable[
        [], tuple[Sequence[object], Mapping[str, object]]
    ]
    | None = None,
    teardown: Callable[..., object] | None = None,
    rounds: int = _DEFAULT_PEDANTIC_ROUNDS,
    warmup_rounds: int = _DEFAULT_WARMUP_ROUNDS,
    iterations: int = _DEFAULT_PEDANTIC_ITERATIONS,
) -> T

Benchmark target with pytest-benchmark pedantic mode.

Source code in src/benchmatrix/bench_harness.py

def pedantic(
    self,
    target: Callable[..., T],
    *,
    args: Sequence[object] | None = None,
    kwargs: Mapping[str, object] | None = None,
    setup: Callable[[], tuple[Sequence[object], Mapping[str, object]]] | None = None,
    teardown: Callable[..., object] | None = None,
    rounds: int = _DEFAULT_PEDANTIC_ROUNDS,
    warmup_rounds: int = _DEFAULT_WARMUP_ROUNDS,
    iterations: int = _DEFAULT_PEDANTIC_ITERATIONS,
) -> T:
    """Benchmark ``target`` with pytest-benchmark pedantic mode."""
    ...

BenchmarkInvocationRecord `dataclass` ¶

Lightweight record returned after one benchmark invocation.

This record is not a timing result. Timing results come from pytest-benchmark's report, saved runs, CSV output, or JSON output.

Attributes:

Name	Type	Description
`metric_name`	`MetricName`	Metric requested for this benchmark invocation.
`implementation_name`	`str`	Name of the implementation under test.
`case_name`	`str`	Name of the input case under test.
`extra_info`	`Mapping[str, object]`	Strict JSON-safe metadata attached to pytest-benchmark output. Values are limited to JSON primitives, lists, and string-keyed mappings after metadata coercion. The metadata includes benchmatrix producer and schema-version markers.

Source code in src/benchmatrix/bench_harness.py

@dataclass(frozen=True, slots=True)
class BenchmarkInvocationRecord:
    """Lightweight record returned after one benchmark invocation.

    This record is not a timing result. Timing results come from
    pytest-benchmark's report, saved runs, CSV output, or JSON output.

    Attributes:
        metric_name: Metric requested for this benchmark invocation.
        implementation_name: Name of the implementation under test.
        case_name: Name of the input case under test.
        extra_info: Strict JSON-safe metadata attached to pytest-benchmark
            output. Values are limited to JSON primitives, lists, and
            string-keyed mappings after metadata coercion. The metadata includes
            benchmatrix producer and schema-version markers.
    """

    metric_name: MetricName
    implementation_name: str
    case_name: str
    extra_info: Mapping[str, object]

ParsedBenchmarkRow `dataclass` ¶

One benchmatrix-tagged row parsed from pytest-benchmark JSON output.

Attributes:

Name	Type	Description
`benchmark_name`	`str`	Name assigned by pytest-benchmark to this benchmark.
`metric_name`	`MetricName`	Benchmatrix metric name from `extra_info`.
`implementation_name`	`str`	Implementation name from `extra_info`.
`case_name`	`str`	Case name from `extra_info`.
`stats`	`_BenchmarkStats`	Raw pytest-benchmark timing statistics.
`extra_info`	`_BenchmarkStats`	Custom metadata from `benchmark.extra_info`.
`derived`	`_BenchmarkStats`	Derived metric-specific statistics computed from JSON output.

Source code in src/benchmatrix/bench_results.py

@dataclass(frozen=True, slots=True)
class ParsedBenchmarkRow:
    """One benchmatrix-tagged row parsed from pytest-benchmark JSON output.

    Attributes:
        benchmark_name: Name assigned by pytest-benchmark to this benchmark.
        metric_name: Benchmatrix metric name from ``extra_info``.
        implementation_name: Implementation name from ``extra_info``.
        case_name: Case name from ``extra_info``.
        stats: Raw pytest-benchmark timing statistics.
        extra_info: Custom metadata from ``benchmark.extra_info``.
        derived: Derived metric-specific statistics computed from JSON output.
    """

    benchmark_name: str
    metric_name: MetricName
    implementation_name: str
    case_name: str
    stats: _BenchmarkStats
    extra_info: _BenchmarkStats
    derived: _BenchmarkStats

BenchmarkJsonError ¶

Bases: BenchmatrixError, ValueError

Raised when pytest-benchmark JSON cannot be parsed as benchmatrix output.

Source code in src/benchmatrix/exceptions.py

class BenchmarkJsonError(BenchmatrixError, ValueError):
    """Raised when pytest-benchmark JSON cannot be parsed as benchmatrix output."""

BenchmatrixError ¶

Bases: Exception

Base class for benchmatrix matrix, metadata, and result errors.

Source code in src/benchmatrix/exceptions.py

class BenchmatrixError(Exception):
    """Base class for benchmatrix matrix, metadata, and result errors."""

MetadataSerializationError ¶

Bases: BenchmatrixError, ValueError

Raised when benchmark metadata cannot be represented as strict JSON.

Source code in src/benchmatrix/exceptions.py

class MetadataSerializationError(BenchmatrixError, ValueError):
    """Raised when benchmark metadata cannot be represented as strict JSON."""

benchmark_batch_throughput ¶

benchmark_batch_throughput(
    benchmark: BenchmarkFixture,
    implementation_name: str,
    function: TargetFunction,
    case_name: str,
    case: BenchmarkCase,
    *,
    config: BenchmarkConfig | None = None,
    stream: TextIO | None = None,
) -> BenchmarkInvocationRecord

Benchmark batch throughput for one implementation and case.

Parameters:

Name	Type	Description	Default
`benchmark`	`BenchmarkFixture`	Pytest-benchmark fixture instance.	required
`implementation_name`	`str`	Name of the implementation under test.	required
`function`	`TargetFunction`	Synchronous function implementation to benchmark. The function must complete the measured work before returning.	required
`case_name`	`str`	Name of the input case under test.	required
`case`	`BenchmarkCase`	Benchmark input case. If `case.work_units` is provided, throughput is later derived as work units per second; otherwise it is derived as calls per second.	required
`config`	`BenchmarkConfig \| None`	Benchmark harness configuration. Defaults to `BenchmarkConfig()`.	`None`
`stream`	`TextIO \| None`	Stream used for progress output. Defaults to `sys.stdout` when progress output is enabled.	`None`

Returns:

Type	Description
`BenchmarkInvocationRecord`	A lightweight invocation record containing metadata attached to the
`BenchmarkInvocationRecord`	benchmark. This is not a timing result.

Raises:

Type	Description
`TypeError`	If `function` is an async function.
`ValueError`	If `case.work_units` is not positive or finite.

Warning

Throughput is derived from one synchronous target invocation. It does not model concurrency, saturation, queueing, or service request load. case.work_units must accurately describe work completed by each target call.

Source code in src/benchmatrix/bench_harness.py

def benchmark_batch_throughput(
    benchmark: BenchmarkFixture,
    implementation_name: str,
    function: TargetFunction,
    case_name: str,
    case: BenchmarkCase,
    *,
    config: BenchmarkConfig | None = None,
    stream: TextIO | None = None,
) -> BenchmarkInvocationRecord:
    """Benchmark batch throughput for one implementation and case.

    Args:
        benchmark: Pytest-benchmark fixture instance.
        implementation_name: Name of the implementation under test.
        function: Synchronous function implementation to benchmark. The
            function must complete the measured work before returning.
        case_name: Name of the input case under test.
        case: Benchmark input case. If ``case.work_units`` is provided,
            throughput is later derived as work units per second; otherwise it
            is derived as calls per second.
        config: Benchmark harness configuration. Defaults to
            ``BenchmarkConfig()``.
        stream: Stream used for progress output. Defaults to ``sys.stdout``
            when progress output is enabled.

    Returns:
        A lightweight invocation record containing metadata attached to the
        benchmark. This is not a timing result.

    Raises:
        TypeError: If ``function`` is an async function.
        ValueError: If ``case.work_units`` is not positive or finite.

    Warning:
        Throughput is derived from one synchronous target invocation. It does
        not model concurrency, saturation, queueing, or service request load.
        ``case.work_units`` must accurately describe work completed by each
        target call.
    """
    resolved_config = _resolve_config(config)
    metric_name = METRIC_BATCH_THROUGHPUT
    extra_info: dict[str, object] = _make_base_extra_info(
        metric_name,
        implementation_name,
        case_name,
        case,
    )
    work_unit_count = case.work_unit_count()

    if work_unit_count is None:
        extra_info[KEY_THROUGHPUT_UNIT] = THROUGHPUT_UNIT_CALLS_PER_SECOND
    else:
        extra_info[KEY_WORK_UNITS] = work_unit_count
        extra_info[KEY_WORK_UNIT_NAME] = case.work_unit_name
        extra_info[KEY_THROUGHPUT_UNIT] = THROUGHPUT_UNIT_WORK_UNITS_PER_SECOND

    final_extra_info = _set_extra_info(benchmark, extra_info)
    _ = _run_target(benchmark, function, case, config=resolved_config, force_pedantic=False)

    record = BenchmarkInvocationRecord(
        metric_name=metric_name,
        implementation_name=implementation_name,
        case_name=case_name,
        extra_info=final_extra_info,
    )
    _maybe_display_invocation_record(record, config=resolved_config, stream=stream)
    return record

benchmark_single_call_latency ¶

benchmark_single_call_latency(
    benchmark: BenchmarkFixture,
    implementation_name: str,
    function: TargetFunction,
    case_name: str,
    case: BenchmarkCase,
    *,
    config: BenchmarkConfig | None = None,
    stream: TextIO | None = None,
) -> BenchmarkInvocationRecord

Benchmark single-call latency for one implementation and case.

Parameters:

Name	Type	Description	Default
`benchmark`	`BenchmarkFixture`	Pytest-benchmark fixture instance.	required
`implementation_name`	`str`	Name of the implementation under test.	required
`function`	`TargetFunction`	Synchronous function implementation to benchmark. The function must complete the measured work before returning.	required
`case_name`	`str`	Name of the input case under test.	required
`case`	`BenchmarkCase`	Benchmark input case.	required
`config`	`BenchmarkConfig \| None`	Benchmark harness configuration. Defaults to `BenchmarkConfig()`.	`None`
`stream`	`TextIO \| None`	Stream used for progress output. Defaults to `sys.stdout` when progress output is enabled.	`None`

Returns:

Type	Description
`BenchmarkInvocationRecord`	A lightweight invocation record containing metadata attached to the
`BenchmarkInvocationRecord`	benchmark. This is not a timing result.

Raises:

Type	Description
`TypeError`	If `function` is an async function.

Warning

This measures completed target-function work only. Input construction, lazy-result consumption, and other setup are excluded unless they occur inside function.

Source code in src/benchmatrix/bench_harness.py

def benchmark_single_call_latency(
    benchmark: BenchmarkFixture,
    implementation_name: str,
    function: TargetFunction,
    case_name: str,
    case: BenchmarkCase,
    *,
    config: BenchmarkConfig | None = None,
    stream: TextIO | None = None,
) -> BenchmarkInvocationRecord:
    """Benchmark single-call latency for one implementation and case.

    Args:
        benchmark: Pytest-benchmark fixture instance.
        implementation_name: Name of the implementation under test.
        function: Synchronous function implementation to benchmark. The
            function must complete the measured work before returning.
        case_name: Name of the input case under test.
        case: Benchmark input case.
        config: Benchmark harness configuration. Defaults to
            ``BenchmarkConfig()``.
        stream: Stream used for progress output. Defaults to ``sys.stdout``
            when progress output is enabled.

    Returns:
        A lightweight invocation record containing metadata attached to the
        benchmark. This is not a timing result.

    Raises:
        TypeError: If ``function`` is an async function.

    Warning:
        This measures completed target-function work only. Input construction,
        lazy-result consumption, and other setup are excluded unless they occur
        inside ``function``.
    """
    resolved_config = _resolve_config(config)
    metric_name = METRIC_SINGLE_CALL_LATENCY
    extra_info: dict[str, object] = _make_base_extra_info(
        metric_name,
        implementation_name,
        case_name,
        case,
    )
    final_extra_info = _set_extra_info(benchmark, extra_info)
    _ = _run_target(benchmark, function, case, config=resolved_config, force_pedantic=False)

    record = BenchmarkInvocationRecord(
        metric_name=metric_name,
        implementation_name=implementation_name,
        case_name=case_name,
        extra_info=final_extra_info,
    )
    _maybe_display_invocation_record(record, config=resolved_config, stream=stream)
    return record

benchmark_tail_latency ¶

benchmark_tail_latency(
    benchmark: BenchmarkFixture,
    implementation_name: str,
    function: TargetFunction,
    case_name: str,
    case: BenchmarkCase,
    *,
    config: BenchmarkConfig | None = None,
    stream: TextIO | None = None,
) -> BenchmarkInvocationRecord

Benchmark latency distribution for one implementation and case.

Parameters:

Name	Type	Description	Default
`benchmark`	`BenchmarkFixture`	Pytest-benchmark fixture instance.	required
`implementation_name`	`str`	Name of the implementation under test.	required
`function`	`TargetFunction`	Synchronous function implementation to benchmark. The function must complete the measured work before returning.	required
`case_name`	`str`	Name of the input case under test.	required
`case`	`BenchmarkCase`	Benchmark input case.	required
`config`	`BenchmarkConfig \| None`	Benchmark harness configuration. Defaults to `BenchmarkConfig()`.	`None`
`stream`	`TextIO \| None`	Stream used for progress output. Defaults to `sys.stdout` when progress output is enabled.	`None`

Returns:

Type	Description
`BenchmarkInvocationRecord`	A lightweight invocation record containing metadata attached to the
`BenchmarkInvocationRecord`	benchmark. This is not a timing result.

Raises:

Type	Description
`TypeError`	If `function` is an async function.

Warning

This uses pedantic mode. Tail percentiles should be calculated from pytest-benchmark JSON data values. This is an implementation-comparison metric, not production p95/p99 latency under load.

If case.fresh_inputs is false and config.pedantic_iterations is greater than one, raw samples should be interpreted as per-round aggregate timings rather than clean one-call latency samples. The harness emits a runtime warning for that configuration.

Source code in src/benchmatrix/bench_harness.py

def benchmark_tail_latency(
    benchmark: BenchmarkFixture,
    implementation_name: str,
    function: TargetFunction,
    case_name: str,
    case: BenchmarkCase,
    *,
    config: BenchmarkConfig | None = None,
    stream: TextIO | None = None,
) -> BenchmarkInvocationRecord:
    """Benchmark latency distribution for one implementation and case.

    Args:
        benchmark: Pytest-benchmark fixture instance.
        implementation_name: Name of the implementation under test.
        function: Synchronous function implementation to benchmark. The
            function must complete the measured work before returning.
        case_name: Name of the input case under test.
        case: Benchmark input case.
        config: Benchmark harness configuration. Defaults to
            ``BenchmarkConfig()``.
        stream: Stream used for progress output. Defaults to ``sys.stdout``
            when progress output is enabled.

    Returns:
        A lightweight invocation record containing metadata attached to the
        benchmark. This is not a timing result.

    Raises:
        TypeError: If ``function`` is an async function.

    Warning:
        This uses pedantic mode. Tail percentiles should be calculated from
        pytest-benchmark JSON ``data`` values. This is an
        implementation-comparison metric, not production p95/p99 latency under
        load.

        If ``case.fresh_inputs`` is false and ``config.pedantic_iterations`` is
        greater than one, raw samples should be interpreted as per-round
        aggregate timings rather than clean one-call latency samples. The
        harness emits a runtime warning for that configuration.
    """
    resolved_config = _resolve_config(config)
    metric_name = METRIC_TAIL_LATENCY
    _warn_for_tail_latency_iteration_semantics(case, resolved_config)

    extra_info: dict[str, object] = _make_base_extra_info(
        metric_name,
        implementation_name,
        case_name,
        case,
    )
    extra_info[KEY_TAIL_LATENCY_NOTE] = (
        "Use pytest-benchmark JSON data to compute p50/p90/p95/p99. "
        "This is not production p95/p99 under load. If pedantic_iterations is "
        "greater than one, samples may represent per-round aggregate timings."
    )
    extra_info[KEY_TAIL_PERCENTILES] = list(TAIL_PERCENTILES)

    final_extra_info = _set_extra_info(benchmark, extra_info)
    _ = _run_target(benchmark, function, case, config=resolved_config, force_pedantic=True)

    record = BenchmarkInvocationRecord(
        metric_name=metric_name,
        implementation_name=implementation_name,
        case_name=case_name,
        extra_info=final_extra_info,
    )
    _maybe_display_invocation_record(record, config=resolved_config, stream=stream)
    return record

deep_copy ¶

deep_copy(value: object) -> object

Return a deep copy of value.

Parameters:

Name	Type	Description	Default
`value`	`object`	Value to copy.	required

Returns:

Type	Description
`object`	A deep copy of `value`.

Source code in src/benchmatrix/bench_harness.py

def deep_copy(value: object) -> object:
    """Return a deep copy of ``value``.

    Args:
        value: Value to copy.

    Returns:
        A deep copy of ``value``.
    """
    return copy.deepcopy(value)

make_benchmark_parameters ¶

make_benchmark_parameters(
    implementations: Mapping[str, TargetFunction],
    cases: Mapping[str, BenchmarkCase]
    | Iterable[BenchmarkCase],
    *,
    metrics: Iterable[MetricName] | None = None,
) -> list[_BenchmarkParameter]

Create pytest parameters for a metric-by-implementation-by-case matrix.

Parameters:

Name	Type	Description	Default
`implementations`	`Mapping[str, TargetFunction]`	Mapping from implementation name to target function.	required
`cases`	`Mapping[str, BenchmarkCase] \| Iterable[BenchmarkCase]`	Mapping or iterable of benchmark input cases.	required
`metrics`	`Iterable[MetricName] \| None`	Metrics to include in the parameter matrix. Defaults to all supported benchmatrix metrics.	`None`

Returns:

Type	Description
`list[_BenchmarkParameter]`	A list of values suitable for `pytest.mark.parametrize`.

Source code in src/benchmatrix/bench_harness.py

def make_benchmark_parameters(
    implementations: Mapping[str, TargetFunction],
    cases: Mapping[str, BenchmarkCase] | Iterable[BenchmarkCase],
    *,
    metrics: Iterable[MetricName] | None = None,
) -> list[_BenchmarkParameter]:
    """Create pytest parameters for a metric-by-implementation-by-case matrix.

    Args:
        implementations: Mapping from implementation name to target function.
        cases: Mapping or iterable of benchmark input cases.
        metrics: Metrics to include in the parameter matrix. Defaults to all
            supported benchmatrix metrics.

    Returns:
        A list of values suitable for ``pytest.mark.parametrize``.
    """
    resolved_metrics = _metric_items(metrics)
    implementation_items = _implementation_items(implementations)
    case_items = _case_items(cases)
    pytest = _load_pytest()
    parameters: list[_BenchmarkParameter] = []

    for metric_name in resolved_metrics:
        for implementation_name, function in implementation_items:
            for case_name, case in case_items:
                parameters.append(
                    pytest.param(
                        metric_name,
                        implementation_name,
                        function,
                        case_name,
                        case,
                        id=f"{metric_name}::{implementation_name}::{case_name}",
                    )
                )

    return parameters

make_benchmark_test ¶

make_benchmark_test(
    implementations: Mapping[str, TargetFunction],
    cases: Mapping[str, BenchmarkCase]
    | Iterable[BenchmarkCase],
    *,
    metrics: Iterable[MetricName] | None = None,
    config: BenchmarkConfig | None = None,
) -> Callable[..., None]

Create a pytest test function for a complete benchmark matrix.

Assign the returned function to a module-level name beginning with test_ so pytest collects it.

Parameters:

Name	Type	Description	Default
`implementations`	`Mapping[str, TargetFunction]`	Mapping from implementation name to target function.	required
`cases`	`Mapping[str, BenchmarkCase] \| Iterable[BenchmarkCase]`	Mapping or iterable of benchmark input cases.	required
`metrics`	`Iterable[MetricName] \| None`	Metrics to include in the parameter matrix. Defaults to all supported benchmatrix metrics.	`None`
`config`	`BenchmarkConfig \| None`	Benchmark harness configuration. Defaults to `BenchmarkConfig()`.	`None`

Returns:

Type	Description
`Callable[..., None]`	A parametrized pytest test function ready for module-level assignment.

Source code in src/benchmatrix/bench_harness.py

def make_benchmark_test(
    implementations: Mapping[str, TargetFunction],
    cases: Mapping[str, BenchmarkCase] | Iterable[BenchmarkCase],
    *,
    metrics: Iterable[MetricName] | None = None,
    config: BenchmarkConfig | None = None,
) -> Callable[..., None]:
    """Create a pytest test function for a complete benchmark matrix.

    Assign the returned function to a module-level name beginning with
    ``test_`` so pytest collects it.

    Args:
        implementations: Mapping from implementation name to target function.
        cases: Mapping or iterable of benchmark input cases.
        metrics: Metrics to include in the parameter matrix. Defaults to all
            supported benchmatrix metrics.
        config: Benchmark harness configuration. Defaults to
            ``BenchmarkConfig()``.

    Returns:
        A parametrized pytest test function ready for module-level assignment.
    """
    resolved_config = _resolve_config(config)
    parameters = make_benchmark_parameters(implementations, cases, metrics=metrics)

    def benchmark_test(
        benchmark: BenchmarkFixture,
        metric_name: MetricName,
        implementation_name: str,
        function: TargetFunction,
        case_name: str,
        case: BenchmarkCase,
    ) -> None:
        """Run one entry in the generated benchmark matrix."""
        _ = run_benchmark_metric(
            benchmark,
            metric_name,
            implementation_name,
            function,
            case_name,
            case,
            config=resolved_config,
        )

    pytest = _load_pytest()
    return pytest.mark.parametrize(
        ("metric_name", "implementation_name", "function", "case_name", "case"),
        parameters,
    )(benchmark_test)

run_benchmark_metric ¶

run_benchmark_metric(
    benchmark: BenchmarkFixture,
    metric_name: MetricName,
    implementation_name: str,
    function: TargetFunction,
    case_name: str,
    case: BenchmarkCase,
    *,
    config: BenchmarkConfig | None = None,
    stream: TextIO | None = None,
) -> BenchmarkInvocationRecord

Run one benchmark metric for one implementation and case.

Parameters:

Name	Type	Description	Default
`benchmark`	`BenchmarkFixture`	Pytest-benchmark fixture instance.	required
`metric_name`	`MetricName`	Metric to benchmark.	required
`implementation_name`	`str`	Name of the implementation under test.	required
`function`	`TargetFunction`	Synchronous function implementation to benchmark.	required
`case_name`	`str`	Name of the input case under test.	required
`case`	`BenchmarkCase`	Benchmark input case.	required
`config`	`BenchmarkConfig \| None`	Benchmark harness configuration. Defaults to `BenchmarkConfig()`.	`None`
`stream`	`TextIO \| None`	Stream used for progress output. Defaults to `sys.stdout` when progress output is enabled.	`None`

Returns:

Type	Description
`BenchmarkInvocationRecord`	A lightweight invocation record containing metadata attached to the
`BenchmarkInvocationRecord`	benchmark. This is not a timing result.

Raises:

Type	Description
`TypeError`	If `function` is an async function.
`ValueError`	If `metric_name` is unsupported.

Source code in src/benchmatrix/bench_harness.py

def run_benchmark_metric(
    benchmark: BenchmarkFixture,
    metric_name: MetricName,
    implementation_name: str,
    function: TargetFunction,
    case_name: str,
    case: BenchmarkCase,
    *,
    config: BenchmarkConfig | None = None,
    stream: TextIO | None = None,
) -> BenchmarkInvocationRecord:
    """Run one benchmark metric for one implementation and case.

    Args:
        benchmark: Pytest-benchmark fixture instance.
        metric_name: Metric to benchmark.
        implementation_name: Name of the implementation under test.
        function: Synchronous function implementation to benchmark.
        case_name: Name of the input case under test.
        case: Benchmark input case.
        config: Benchmark harness configuration. Defaults to
            ``BenchmarkConfig()``.
        stream: Stream used for progress output. Defaults to ``sys.stdout``
            when progress output is enabled.

    Returns:
        A lightweight invocation record containing metadata attached to the
        benchmark. This is not a timing result.

    Raises:
        TypeError: If ``function`` is an async function.
        ValueError: If ``metric_name`` is unsupported.
    """
    resolved_config = _resolve_config(config)
    resolved_metric_name = _validate_metric_name(metric_name)

    if resolved_metric_name == METRIC_SINGLE_CALL_LATENCY:
        return benchmark_single_call_latency(
            benchmark,
            implementation_name,
            function,
            case_name,
            case,
            config=resolved_config,
            stream=stream,
        )

    if resolved_metric_name == METRIC_BATCH_THROUGHPUT:
        return benchmark_batch_throughput(
            benchmark,
            implementation_name,
            function,
            case_name,
            case,
            config=resolved_config,
            stream=stream,
        )

    if resolved_metric_name == METRIC_TAIL_LATENCY:
        return benchmark_tail_latency(
            benchmark,
            implementation_name,
            function,
            case_name,
            case,
            config=resolved_config,
            stream=stream,
        )

    raise ValueError(f"Unsupported benchmark metric: {resolved_metric_name!r}")

shallow_copy ¶

shallow_copy(value: object) -> object

Return a shallow copy of value.

Parameters:

Name	Type	Description	Default
`value`	`object`	Value to copy.	required

Returns:

Type	Description
`object`	A shallow copy of `value`.

Source code in src/benchmatrix/bench_harness.py

def shallow_copy(value: object) -> object:
    """Return a shallow copy of ``value``.

    Args:
        value: Value to copy.

    Returns:
        A shallow copy of ``value``.
    """
    return copy.copy(value)

display_benchmark_row ¶

display_benchmark_row(
    row: ParsedBenchmarkRow, stream: TextIO | None = None
) -> None

Print one metric-aware summary of a parsed benchmark row.

Parameters:

Name	Type	Description	Default
`row`	`ParsedBenchmarkRow`	Parsed benchmark row to display.	required
`stream`	`TextIO \| None`	Output stream. Defaults to `sys.stdout`.	`None`

Source code in src/benchmatrix/bench_results.py

def display_benchmark_row(
    row: ParsedBenchmarkRow,
    stream: TextIO | None = None,
) -> None:
    """Print one metric-aware summary of a parsed benchmark row.

    Args:
        row: Parsed benchmark row to display.
        stream: Output stream. Defaults to ``sys.stdout``.
    """
    output = sys.stdout if stream is None else stream
    prefix = f"[{row.metric_name}] implementation={row.implementation_name} case={row.case_name}"

    if row.metric_name == METRIC_SINGLE_CALL_LATENCY:
        message = (
            f"{prefix} mean={_format_seconds(row.stats.get(STAT_MEAN))} "
            + f"median={_format_seconds(row.stats.get(STAT_MEDIAN))} "
            + f"min={_format_seconds(row.stats.get(STAT_MIN))}"
        )
        print(
            message,
            file=output,
        )
        return

    if row.metric_name == METRIC_BATCH_THROUGHPUT:
        message = (
            f"{prefix} "
            + f"throughput_mean={_format_rate(row.derived.get(DERIVED_THROUGHPUT_MEAN))} "
            + f"throughput_median={_format_rate(row.derived.get(DERIVED_THROUGHPUT_MEDIAN))} "
            + f"unit={row.derived.get(DERIVED_THROUGHPUT_UNIT_LABEL)}"
        )
        print(
            message,
            file=output,
        )
        return

    if row.metric_name == METRIC_TAIL_LATENCY:
        message = (
            f"{prefix} p50={_format_seconds(row.derived.get(DERIVED_P50))} "
            + f"p95={_format_seconds(row.derived.get(DERIVED_P95))} "
            + f"p99={_format_seconds(row.derived.get(DERIVED_P99))} "
            + f"max={_format_seconds(row.derived.get(DERIVED_MAX))}"
        )
        print(
            message,
            file=output,
        )
        return

    print(
        f"{prefix} mean={_format_seconds(row.stats.get(STAT_MEAN))}",
        file=output,
    )

display_benchmark_rows ¶

display_benchmark_rows(
    rows: Iterable[ParsedBenchmarkRow],
    stream: TextIO | None = None,
) -> None

Print concise metric-aware summaries of parsed benchmark rows.

Parameters:

Name	Type	Description	Default
`rows`	`Iterable[ParsedBenchmarkRow]`	Parsed benchmark rows.	required
`stream`	`TextIO \| None`	Output stream. Defaults to `sys.stdout`.	`None`

Source code in src/benchmatrix/bench_results.py

def display_benchmark_rows(
    rows: Iterable[ParsedBenchmarkRow],
    stream: TextIO | None = None,
) -> None:
    """Print concise metric-aware summaries of parsed benchmark rows.

    Args:
        rows: Parsed benchmark rows.
        stream: Output stream. Defaults to ``sys.stdout``.
    """
    for row in rows:
        display_benchmark_row(row, stream=stream)

load_benchmark_json ¶

load_benchmark_json(
    path: str | Path,
) -> list[ParsedBenchmarkRow]

Load benchmatrix-tagged pytest-benchmark JSON and derive metric views.

Parameters:

Name	Type	Description	Default
`path`	`str \| Path`	Path to a JSON file created with `--benchmark-json`.	required

Returns:

Type	Description
`list[ParsedBenchmarkRow]`	Benchmatrix-tagged rows with raw pytest-benchmark statistics and derived
`list[ParsedBenchmarkRow]`	metric-specific fields. Non-benchmatrix rows are rejected.

Raises:

Type	Description
`BenchmarkJsonError`	If the JSON does not have the expected pytest-benchmark and benchmatrix structure.

Source code in src/benchmatrix/bench_results.py

def load_benchmark_json(path: str | Path) -> list[ParsedBenchmarkRow]:
    """Load benchmatrix-tagged pytest-benchmark JSON and derive metric views.

    Args:
        path: Path to a JSON file created with ``--benchmark-json``.

    Returns:
        Benchmatrix-tagged rows with raw pytest-benchmark statistics and derived
        metric-specific fields. Non-benchmatrix rows are rejected.

    Raises:
        BenchmarkJsonError: If the JSON does not have the expected
            pytest-benchmark and benchmatrix structure.
    """
    path_obj = Path(path)

    try:
        payload = _load_json(path_obj)
    except OSError as exc:
        raise BenchmarkJsonError(f"Could not read benchmark JSON: {path_obj}") from exc
    except json.JSONDecodeError as exc:
        raise BenchmarkJsonError(f"Invalid JSON in benchmark file: {path_obj}") from exc

    payload_mapping = _require_mapping(payload, path="root")
    benchmarks = _require_list(
        payload_mapping.get(JSON_KEY_BENCHMARKS),
        path=f"root.{JSON_KEY_BENCHMARKS}",
    )

    rows: list[ParsedBenchmarkRow] = []
    for index, benchmark_entry in enumerate(benchmarks):
        entry_path = f"root.{JSON_KEY_BENCHMARKS}[{index}]"
        entry = _require_mapping(benchmark_entry, path=entry_path)
        extra_info = _require_mapping(
            entry.get(JSON_KEY_EXTRA_INFO),
            path=f"{entry_path}.{JSON_KEY_EXTRA_INFO}",
        )
        _require_benchmatrix_schema(extra_info, path=f"{entry_path}.{JSON_KEY_EXTRA_INFO}")

        stats = _require_mapping(
            entry.get(JSON_KEY_STATS),
            path=f"{entry_path}.{JSON_KEY_STATS}",
        )

        metric_name = _require_metric_name(
            extra_info.get(KEY_METRIC_NAME),
            path=f"{entry_path}.{JSON_KEY_EXTRA_INFO}.{KEY_METRIC_NAME}",
        )
        data = _extract_benchmark_data(entry, stats, metric_name, path=entry_path)

        rows.append(
            ParsedBenchmarkRow(
                benchmark_name=_benchmark_name(entry),
                metric_name=metric_name,
                implementation_name=_require_string(
                    extra_info.get(KEY_IMPLEMENTATION_NAME),
                    path=f"{entry_path}.{JSON_KEY_EXTRA_INFO}.{KEY_IMPLEMENTATION_NAME}",
                ),
                case_name=_require_string(
                    extra_info.get(KEY_CASE_NAME),
                    path=f"{entry_path}.{JSON_KEY_EXTRA_INFO}.{KEY_CASE_NAME}",
                ),
                stats=stats,
                extra_info=extra_info,
                derived=_derive_stats(metric_name, stats, extra_info, data),
            )
        )

    return rows

benchmatrix¶

benchmatrix ¶

TargetFunction module-attribute ¶

BenchmarkCase dataclass ¶

__post_init__ ¶

make_call ¶

work_unit_count ¶

from_values classmethod ¶

BenchmarkConfig dataclass ¶

__post_init__ ¶

BenchmarkFixture ¶

__call__ ¶

pedantic ¶

BenchmarkInvocationRecord dataclass ¶

ParsedBenchmarkRow dataclass ¶

BenchmarkJsonError ¶

BenchmatrixError ¶

MetadataSerializationError ¶

benchmark_batch_throughput ¶

benchmark_single_call_latency ¶

benchmark_tail_latency ¶

deep_copy ¶

make_benchmark_parameters ¶

make_benchmark_test ¶

run_benchmark_metric ¶

shallow_copy ¶

display_benchmark_row ¶

display_benchmark_rows ¶

load_benchmark_json ¶

TargetFunction `module-attribute` ¶

BenchmarkCase `dataclass` ¶

from_values `classmethod` ¶

BenchmarkConfig `dataclass` ¶

call ¶

BenchmarkInvocationRecord `dataclass` ¶

ParsedBenchmarkRow `dataclass` ¶