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.

In the 2.5.0 Python uniqueness path, unique_keys compares hashable and unhashable keys across categories, keeping the first accepted equal key and its value. Cross-category comparisons may run user equality callbacks and take O(H × U) time for H hashable and U unhashable keys. The operation retains distinct keys; to_dict still requires every emitted key to be hashable.

In the 2.5.0 Python path, collect_values and aggregate_values propagate errors raised by a key's custom hash or equality operation unchanged. An unhashable key raises TypeError.

For a streaming collector, collect_values creates one state per key and skips further step calls for that key after its done predicate is true. Other keys can still receive values. A state of None remains a state and does not cause another initialization. This differs from Collectors.grouping_by, which offers every item to its downstream collector regardless of done.

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 was added in 2.2.0.

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, S, R]) -> dict[K, R]

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

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: Collector[Any, Any, Any],
) -> dict[K, dict[str, Any]]

Run named collectors independently for every key.

Parameters:

Name	Type	Description	Default
**aggregations	Collector[Any, Any, Any]	Result names mapped to collectors 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.