spdl.pipeline.defs.PathVariants

PathVariants(router: Callable[[Any], int] | Callable[[Any], Awaitable[int]], paths: Sequence[Sequence[PipeConfig[Any, Any] | AggregateConfig[Any] | DisaggregateConfig[Any] | PathVariantsConfig[Any]]], name: str | None = None, *, batched: Literal[False] = False) → PathVariantsConfig[Any]

PathVariants(router: Callable[[Sequence[Any]], Sequence[int]] | Callable[[Sequence[Any]], Awaitable[Sequence[int]]], paths: Sequence[Sequence[PipeConfig[Any, Any] | AggregateConfig[Any] | DisaggregateConfig[Any] | PathVariantsConfig[Any]]], name: str | None = None, *, batched: Literal[True]) → PathVariantsConfig[Any]

Create a PathVariantsConfig for variant path routing.

See also

Example: Pipeline definitions

Illustrates how to build a complex pipeline.

Routes each incoming item to one of several processing paths based on a router function. Each path is an independent chain of pipe configs. The outputs of all paths are merged back into a single stream.

This is useful when items need different processing depending on runtime conditions — e.g., routing cached items to a fast cache-read path while uncached items go through full data loading, or splitting between local and remote processing.

With batched=True the routing is applied per batch instead of per item: the router receives a whole batch (a list) and returns one path index per element, the batch is partitioned into per-path sub-batches (each path’s ops then operate on a list), and the sub-batches are concatenated back into one batch by the merge. This amortizes the per-item routing and fan-out/fan-in overhead over a whole batch — aggregate the source into batches upstream of the stage. Each path op receives and returns a list; a path may drop elements by returning a shorter list, and must tolerate an empty input list (a path that received no elements for a given batch).

Changed in version 0.6.0: Fixed to work with a continuous source.

Added in version 0.6.0: The batched argument.

Parameters:

• router – A callable 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 a sequence of per-element int indices (one per item, same length as the batch). Every index must be in range [0, len(paths)).

• paths – A sequence of paths. Each path is a sequence of pipe configs (PipeConfig, AggregateConfig, DisaggregateConfig, or nested PathVariantsConfig). SourceConfig and SinkConfig are not allowed.

• name – Optional name for the stage.

• batched – If True, route whole batches instead of single items (see above).

Returns:

The config object.

Raises:

ValueError – If router is not callable, paths is empty, or a path contains SourceConfig or SinkConfig.

Example:

from spdl.pipeline.defs import PathVariants, Pipe, PipelineConfig, SourceConfig, SinkConfig

config = PipelineConfig(
    src=SourceConfig(items),
    pipes=[
        PathVariants(
            router=lambda item: 0 if item in cache else 1,
            paths=[
                [Pipe(load_from_cache)],   # path 0: cache hit
                [Pipe(load_from_source)],  # path 1: cache miss
            ],
        ),
    ],
    sink=SinkConfig(buffer_size=10),
)

Batched routing (aggregate first; each path op takes and returns a list):

config = PipelineConfig(
    src=SourceConfig(items),
    pipes=[
        Aggregate(64),
        PathVariants(
            router=lambda batch: [0 if x in cache else 1 for x in batch],
            paths=[
                [Pipe(load_batch_from_cache)],   # path 0: cache hits
                [Pipe(load_batch_from_source)],  # path 1: cache misses
            ],
            batched=True,
        ),
    ],
    sink=SinkConfig(buffer_size=10),
)