Pairs

Pairs[K, V] is a key/value view over a lazy flow of two-tuples. It provides key-, value-, and pair-aware transforms plus per-key collection and aggregation.

with_engine() keeps the Pairs view while changing the underlying Flow policy. to_flow() returns that Flow without copying values. Inspect the returned Flow when it will be consumed as a Flow; pair terminals such as to_dict() may use a pair-specific direct path and do not expose a separate explain() API.

Use run_with_report() to execute to_dict, group_values, collect_values, or aggregate_values and inspect the route. The terminal runs once; extra arguments are forwarded to it. The result contains the terminal's value and an immutable execution report. This reporting method is available in the current source tree and has not yet been released on PyPI.

from fpstreams import flow

result = flow([("red", 2), ("red", 3)]).pairs().run_with_report("group_values")
assert result.value == {"red": [2, 3]}

fpstreams.Pairs

Bases: Generic[K, V]

A lazy key-value view backed by a synchronous Flow.

to_flow

to_flow() -> Flow[tuple[K, V]]

Return the underlying flow of key/value tuples.

Returns:

Type	Description
Flow[tuple[K, V]]	The same underlying Flow; no copy or additional operation is created.

with_engine

with_engine(engine: Engine) -> Pairs[K, V]

Return an equivalent pair view requesting one Flow execution engine.

Parameters:

Name	Type	Description	Default
engine	Engine	"auto", "python", or "native".	required

Returns:

Type	Description
Pairs[K, V]	Lazy Pairs backed by the engine-adjusted Flow.

run_with_report

run_with_report(
    terminal: str, /, *args: Any, **kwargs: Any
) -> ExecutionResult[Any]

Execute one pair terminal and return its value with query-owned metrics.

Parameters:

Name	Type	Description	Default
terminal	str	to_dict, group_values, collect_values, or aggregate_values.	required
*args	Any	Positional arguments passed to that terminal.	()
**kwargs	Any	Keyword arguments passed to that terminal.	{}

Returns:

Type	Description
ExecutionResult[Any]	The terminal value and an immutable report. The source is consumed once.

Raises:

Type	Description
ValueError	If the name is not a reportable pair terminal.

keys

keys() -> Flow[K]

Select only keys as a Flow.

Returns:

Type	Description
Flow[K]	A flow containing the first element of every pair.

values

values() -> Flow[V]

Select only values as a Flow.

Returns:

Type	Description
Flow[V]	A flow containing the second element of every pair.

map_pairs

map_pairs(
    function: Callable[[K, V], tuple[R, U]],
) -> Pairs[R, U]

Transform each key/value pair into a new pair.

Parameters:

Name	Type	Description	Default
function	Callable[[K, V], tuple[R, U]]	Called as function(key, value) and returns the replacement pair.	required

Returns:

Type	Description
Pairs[R, U]	A lazy pair pipeline containing each returned (new_key, new_value).

flat_map_pairs

flat_map_pairs(
    function: Callable[[K, V], Iterable[tuple[R, U]]],
) -> Pairs[R, U]

Transform each pair into zero or more pairs.

Parameters:

Name	Type	Description	Default
function	Callable[[K, V], Iterable[tuple[R, U]]]	Called as function(key, value) and returns an iterable of replacement pairs.	required

Returns:

Type	Description
Pairs[R, U]	A lazy pair pipeline that emits every returned iterable in source order.

filter_pairs

filter_pairs(
    predicate: Callable[[K, V], bool] | RowExpr,
) -> Pairs[K, V]

Keep pairs for which predicate returns true.

Parameters:

Name	Type	Description	Default
predicate	Callable[[K, V], bool] | RowExpr	A callable invoked as predicate(key, value), or a RowExpr evaluated against the complete pair so col(0) selects the key and col(1) the value.	required

Returns:

Type	Description
Pairs[K, V]	A lazy pair pipeline containing only pairs with truthy predicate results.

tap

tap(action: Callable[[K, V], object]) -> Pairs[K, V]

Run a side effect for each pair while passing the pair through.

Parameters:

Name	Type	Description	Default
action	Callable[[K, V], object]	Called as action(key, value) before the original pair is emitted.	required

Returns:

Type	Description
Pairs[K, V]	A lazy pair pipeline that passes every original pair through unchanged.

take

take(count: int) -> Pairs[K, V]

Emit at most count pairs.

Parameters:

Name	Type	Description	Default
count	int	Maximum number of leading pairs to emit.	required

Returns:

Type	Description
Pairs[K, V]	A lazy pair pipeline containing only the first count pairs.

drop

drop(count: int) -> Pairs[K, V]

Skip count pairs before yielding the remainder.

Parameters:

Name	Type	Description	Default
count	int	Number of leading pairs to consume without emitting.	required

Returns:

Type	Description
Pairs[K, V]	A lazy pair pipeline containing every pair after the first count.

sort_by_key

sort_by_key(*, reverse: bool = False) -> Pairs[K, V]

Sort pairs by key.

Parameters:

Name	Type	Description	Default
reverse	bool	Sort keys descending when true.	False

Returns:

Type	Description
Pairs[K, V]	A lazy pair pipeline globally ordered by key.

sort_by_value

sort_by_value(*, reverse: bool = False) -> Pairs[K, V]

Sort pairs by value.

Parameters:

Name	Type	Description	Default
reverse	bool	Sort values descending when true.	False

Returns:

Type	Description
Pairs[K, V]	A lazy pair pipeline globally ordered by value.

unique_keys

unique_keys() -> Pairs[K, V]

Keep the first pair for each key.

Returns:

Type	Description
Pairs[K, V]	A lazy pair pipeline containing the earliest pair for each distinct key.

map_keys

map_keys(function: Callable[[K], R]) -> Pairs[R, V]

Transform keys while leaving values unchanged.

Parameters:

Name	Type	Description	Default
function	Callable[[K], R]	Maps each key to its replacement key.	required

Returns:

Type	Description
Pairs[R, V]	A lazy pair pipeline of (function(key), value) pairs.

map_values

map_values(function: Callable[[V], R]) -> Pairs[K, R]

Transform values while leaving keys unchanged.

Parameters:

Name	Type	Description	Default
function	Callable[[V], R]	Maps each value to its replacement value.	required

Returns:

Type	Description
Pairs[K, R]	A lazy pair pipeline of (key, function(value)) pairs.

filter_keys

filter_keys(predicate: Callable[[K], bool]) -> Pairs[K, V]

Keep pairs whose key satisfies predicate.

Parameters:

Name	Type	Description	Default
predicate	Callable[[K], bool]	Called with each key to decide whether its pair is retained.	required

Returns:

Type	Description
Pairs[K, V]	A lazy pair pipeline containing pairs whose keys satisfy predicate.

filter_values

filter_values(
    predicate: Callable[[V], bool],
) -> Pairs[K, V]

Keep pairs whose value satisfies predicate.

Parameters:

Name	Type	Description	Default
predicate	Callable[[V], bool]	Called with each value to decide whether its pair is retained.	required

Returns:

Type	Description
Pairs[K, V]	A lazy pair pipeline containing pairs whose values satisfy predicate.

invert

invert() -> Pairs[V, K]

Swap the key and value in every pair.

Returns:

Type	Description
Pairs[V, K]	A lazy pair pipeline containing (value, key) for each source pair.

to_dict

to_dict(
    *,
    on_duplicate: Literal[
        "error", "first", "last"
    ] = "error",
) -> dict[K, V]

Collect pairs into a dictionary using the duplicate-key policy.

Parameters:

Name	Type	Description	Default
on_duplicate	Literal['error', 'first', 'last']	error raises, first keeps the earliest value, and last keeps the latest value for a repeated key.	'error'

Returns:

Type	Description
dict[K, V]	One selected value for each hashable key, in first-key encounter order.

Raises:

Type	Description
DuplicateKeyError	If a key repeats under the error policy.
ValueError	If on_duplicate is not error, first, or last.

group_values

group_values() -> dict[K, list[V]]

Collect all values for each key in encounter order.

Returns:

Type	Description
dict[K, list[V]]	Lists of values keyed by each hashable key, preserving key and value encounter order.

collect_values

collect_values(
    collector: Collector[V, Any, R]
    | Callable[[Iterable[V]], R],
) -> dict[K, R]

Run one Collector or callable independently for every key.

Parameters:

Name	Type	Description	Default
collector	Collector[V, Any, R] | Callable[[Iterable[V]], R]	A streaming Collector applied independently per key, or a callable that receives that key's complete value list.	required

Returns:

Type	Description
dict[K, R]	Each hashable key mapped to its independently finished collector result.

aggregate_values

aggregate_values(
    **aggregations: Aggregator,
) -> dict[K, dict[str, Any]]

Run named Aggregators independently for every key.

Parameters:

Name	Type	Description	Default
**aggregations	Aggregator	Result names mapped to aggregators maintained independently per key.	{}

Returns:

Type	Description
dict[K, dict[str, Any]]	Each hashable key mapped to a dictionary of its named finished aggregation values.