spdl.pipeline.defs.PathVariantsConfig

class PathVariantsConfig(name: str, router: Callable[[Any], int] | Callable[[Any], Awaitable[int]] | Callable[[Sequence[Any]], Sequence[int]] | Callable[[Sequence[Any]], Awaitable[Sequence[int]]], paths: tuple[tuple[PipeConfig[Any, Any] | AggregateConfig[Any] | DisaggregateConfig[Any] | PathVariantsConfig[Any], ...], ...], batched: bool = False)

Configuration for variant path routing.

Use PathVariants() to create a config.

Routes each incoming item to one of several processing paths based on a router function. All paths produce the same output type and their outputs are merged back into a single stream.

This is useful when items need different processing depending on runtime conditions. For example:

• Caching: route cached items to a fast cache-read path while uncached items go through full data loading.

• Hybrid processing: split items between local and remote processing based on size or availability.

See also

Example: Pipeline definitions

Illustrates how to build a complex pipeline.

Example:

from spdl.pipeline import PipelineBuilder

def cache_router(item):
    return 0 if item in cache else 1

pipeline = (
    PipelineBuilder()
    .add_source(items)
    .path_variants(
        router=cache_router,
        paths=[
            [Pipe(load_from_cache)],   # path 0: cache hit
            [Pipe(load_from_source)],  # path 1: cache miss
        ],
    )
    .add_sink(buffer_size=10)
    .build()
)

Note

Paths can only contain pipe stages (PipeConfig, AggregateConfig, DisaggregateConfig, or nested PathVariantsConfig). SourceConfig and SinkConfig are not allowed in paths.

Attributes

batched	If True, route whole batches instead of single items.
name	Name of the path variants stage.
router	A function that selects the path for each input.
paths	Alternative processing paths.

batched: bool = False

If True, route whole batches instead of single items.

Each incoming item is a batch (a list). router(batch) returns one path index per element; the batch is partitioned into per-path sub-batches (lists, order preserved), each path processes its sub-batch as a list, and the merge concatenates the sub-batches back into one batch. This amortizes the routing and fan-out/fan-in overhead over a whole batch instead of paying it per item. Aggregate the source into batches upstream of this stage when using it.

name: str

Name of the path variants stage.

paths: tuple[tuple[PipeConfig[Any, Any] | AggregateConfig[Any] | DisaggregateConfig[Any] | PathVariantsConfig[Any], ...], ...]

Alternative processing paths. Each path is a tuple of pipe configs.

router: Callable[[Any], int] | Callable[[Any], Awaitable[int]] | Callable[[Sequence[Any]], Sequence[int]] | Callable[[Sequence[Any]], Awaitable[Sequence[int]]]

A function that selects the path for each input.

In per-item mode (default) it takes an item and returns an int index. In batched mode it takes a batch (list) and returns one int index per element. Either form can be a regular function or an async function/callable.