Collecting and aggregation

Collectors reduce a flow into a result. Aggregators are composable state machines used by named and grouped aggregation.

Collector

fpstreams.Collector `dataclass`

Bases: Generic[T, S, R]

Describe an immutable streaming reduction from input items to a result.

initializer creates independent state, step consumes one item, and finish converts final state to the public result. An optional combine merges partial states. done enables source short-circuiting, and native carries planner metadata without changing Python execution.

Built-in collectors

fpstreams.Collectors

Bases: Generic[T]

Build reusable collectors for containers, grouping, adaptation, and summaries.

to_list `staticmethod`

to_list() -> Collector[T, list[T], list[T]]

Build an order-preserving, mergeable collector for all input items.

Returns:

Type	Description
`Collector[T, list[T], list[T]]`	A reducer whose finished list is also its accumulated state.

to_set `staticmethod`

to_set() -> Collector[T, set[T], set[T]]

Build a collector that retains each distinct hashable input item.

Returns:

Type	Description
`Collector[T, set[T], set[T]]`	A collector with set-union support for partial states.

to_tuple `staticmethod`

to_tuple() -> Collector[T, list[T], tuple[T, ...]]

Build an order-preserving collector that finishes list state as a tuple.

Returns:

Type	Description
`Collector[T, list[T], tuple[T, ...]]`	A mergeable reducer returning an immutable tuple.

joining `staticmethod`

joining(
    delimiter: str = "",
) -> Collector[Any, list[str], str]

Build an order-preserving collector that joins each item's str value.

Conversion occurs as items are stepped. Empty input finishes as an empty string.

Parameters:

Name	Type	Description	Default
`delimiter`	`str`	The string inserted between collected values.	`''`

Returns:

Type	Description
`Collector[Any, list[str], str]`	A mergeable reducer that joins its accumulated strings with `delimiter`.

grouping_by `staticmethod`

grouping_by(
    classifier: Selector,
    downstream: Collector[T, Any, R]
    | Callable[[Iterable[T]], R]
    | None = None,
) -> Collector[T, dict[Any, Any], dict[Any, R | list[T]]]

Build a collector that maintains one downstream state per selected key.

Only encountered keys appear in the result, in first-encounter order. Each item is stepped into its group even if the downstream collector has an early-completion predicate. A callable downstream is invoked at finish time with a materialized list for each group; omitting it collects group members into lists.

Parameters:

Name	Type	Description	Default
`classifier`	`Selector`	Selector producing a hashable group key for each item.	required
`downstream`	`Collector[T, Any, R] \| Callable[[Iterable[T]], R] \| None`	Per-group collector or callable, defaulting to list collection.	`None`

Returns:

Type	Description
`Collector[T, dict[Any, Any], dict[Any, R \| list[T]]]`	A collector that finishes every encountered group's state into a dictionary.

partitioning_by `staticmethod`

partitioning_by(
    predicate: Selector,
    downstream: Collector[T, Any, R]
    | Callable[[Iterable[T]], R]
    | None = None,
) -> Collector[T, dict[bool, Any], dict[bool, R | list[T]]]

Build a collector with independently reduced false and true partitions.

Both Boolean keys are always present, even when one or both partitions receive no items. Each source item is stepped into one partition without consulting the downstream collector's early-completion predicate. Omitting downstream collects partition members into lists.

Parameters:

Name	Type	Description	Default
`predicate`	`Selector`	Selector whose truth value chooses the partition.	required
`downstream`	`Collector[T, Any, R] \| Callable[[Iterable[T]], R] \| None`	Collector or callable applied independently to both partitions.	`None`

Returns:

Type	Description
`Collector[T, dict[bool, Any], dict[bool, R \| list[T]]]`	A collector returning `{False: false_result, True: true_result}`.

mapping `staticmethod`

mapping(
    mapper: Selector,
    downstream: Collector[U, Any, R]
    | Callable[[Iterable[U]], R],
) -> Collector[T, Any, R]

Build a collector that selects a new value before each downstream step.

The adapter preserves downstream initialization, finishing, combining, and early completion, but does not carry downstream native metadata or reducer-law type.

Parameters:

Name	Type	Description	Default
`mapper`	`Selector`	Selector applied once to each consumed source item.	required
`downstream`	`Collector[U, Any, R] \| Callable[[Iterable[U]], R]`	Collector or callable receiving mapped values.	required

Returns:

Type	Description
`Collector[T, Any, R]`	A collector that maps each source item before returning the downstream result.

filtering `staticmethod`

filtering(
    predicate: Selector,
    downstream: Collector[T, Any, R]
    | Callable[[Iterable[T]], R],
) -> Collector[T, Any, R]

Build a collector that steps downstream only for predicate matches.

Rejected items leave downstream state unchanged. Initialization, finishing, combining, and early completion are preserved; native metadata and reducer-law type are not.

Parameters:

Name	Type	Description	Default
`predicate`	`Selector`	Selector evaluated for truth against each item.	required
`downstream`	`Collector[T, Any, R] \| Callable[[Iterable[T]], R]`	Collector or callable receiving matching items.	required

Returns:

Type	Description
`Collector[T, Any, R]`	A collector that filters source items before returning the downstream result.

flat_mapping `staticmethod`

flat_mapping(
    mapper: Selector,
    downstream: Collector[U, Any, R]
    | Callable[[Iterable[U]], R],
) -> Collector[T, Any, R]

Build a collector that steps downstream over each item's expanded iterable.

Expansion stops as soon as the downstream state reports completion. A nested iterator exposing close is closed after exhaustion, early completion, or error. Downstream combining and early completion are preserved, while native metadata and reducer-law type are not.

Parameters:

Name	Type	Description	Default
`mapper`	`Selector`	Selector returning an iterable for each source item.	required
`downstream`	`Collector[U, Any, R] \| Callable[[Iterable[U]], R]`	Collector or callable receiving nested items.	required

Returns:

Type	Description
`Collector[T, Any, R]`	A collector that flattens source items before returning the downstream result.

collecting_and_then `staticmethod`

collecting_and_then(
    downstream: Collector[T, Any, R]
    | Callable[[Iterable[T]], R],
    finisher: Callable[[R], U],
) -> Collector[T, Any, U]

Build a collector that applies one more transformation after downstream finish.

The downstream state machine, combiner, and early-completion behavior are preserved. finisher is called exactly once with the downstream public result.

Parameters:

Name	Type	Description	Default
`downstream`	`Collector[T, Any, R] \| Callable[[Iterable[T]], R]`	Collector or callable that produces the intermediate result.	required
`finisher`	`Callable[[R], U]`	Callable converting that intermediate result to the final value.	required

Returns:

Type	Description
`Collector[T, Any, U]`	An adapted collector returning `finisher(downstream_result)`.

Raises:

Type	Description
`TypeError`	If `finisher` is not callable.

teeing `staticmethod`

teeing(
    left: Collector[T, Any, R] | Callable[[Iterable[T]], R],
    right: Collector[T, Any, U]
    | Callable[[Iterable[T]], U],
    merger: Callable[[R, U], V],
) -> Collector[T, _TeeState, V]

Build a collector that shares one source between two downstream collectors.

Each input is offered only to downstream states that are not already complete. Source consumption stops when both are complete, then merger receives their finished results in left-to-right order. Callable downstreams buffer their respective items.

Parameters:

Name	Type	Description	Default
`left`	`Collector[T, Any, R] \| Callable[[Iterable[T]], R]`	First collector or iterable-consuming callable.	required
`right`	`Collector[T, Any, U] \| Callable[[Iterable[T]], U]`	Second collector or iterable-consuming callable.	required
`merger`	`Callable[[R, U], V]`	A callable that merges two downstream results.	required

Returns:

Type	Description
`Collector[T, _TeeState, V]`	A short-circuiting collector returning the merger result.

Raises:

Type	Description
`TypeError`	If either downstream or `merger` is not callable as required.

counting `staticmethod`

counting() -> Collector[T, int, int]

Build a constant-state reducer that increments once per input item.

Returns:

Type	Description
`Collector[T, int, int]`	A commutative reducer returning zero for empty input.

summing `staticmethod`

summing(
    selector: Selector | None = None,
) -> Collector[T, Any, Any]

Build a collector that adds selected values from a zero identity.

Parameters:

Name	Type	Description	Default
`selector`	`Selector \| None`	Value selector; `None` adds each whole input item.	`None`

Returns:

Type	Description
`Collector[T, Any, Any]`	A collector returning zero for empty input and supporting partial-state addition.

averaging `staticmethod`

averaging(
    selector: Selector | None = None,
) -> Collector[T, tuple[Any, int], float]

Build a collector that tracks selected-value sum and count for a mean.

Empty input returns 0.0. Nonempty input divides the accumulated sum by count during finishing, and partial (sum, count) states can be combined component-wise.

Parameters:

Name	Type	Description	Default
`selector`	`Selector \| None`	Value selector; `None` averages each whole input item.	`None`

Returns:

Type	Description
`Collector[T, tuple[Any, int], float]`	A collector returning the arithmetic mean or `0.0` for no values.

summarizing `staticmethod`

summarizing(
    selector: Selector | None = None,
) -> Collector[T, SummaryStatistics, SummaryStatistics]

Build a collector returning mutable count, sum, extrema, and average state.

Selected values update SummaryStatistics directly. Empty input changes the minimum and maximum from infinities to 0.0; the sum, count, and derived average are already zero. The returned object is the final mutable state.

Parameters:

Name	Type	Description	Default
`selector`	`Selector \| None`	Value selector; `None` summarizes each whole input item.	`None`

Returns:

Type	Description
`Collector[T, SummaryStatistics, SummaryStatistics]`	A one-pass collector returning SummaryStatistics.

first `staticmethod`

first() -> Collector[T, Any, T | None]

Build a collector that stops after and returns the first input item.

Empty input returns None. A first item whose value is itself None is still treated as a completed result.

Returns:

Type	Description
`Collector[T, Any, T \| None]`	A collector that consumes at most one item.

last `staticmethod`

last() -> Collector[T, Any, T | None]

Build a collector that consumes all input and returns its last item.

Empty input returns None.

Returns:

Type	Description
`Collector[T, Any, T \| None]`	A collector retaining only the most recently consumed item.

head `staticmethod`

head(count: int) -> Collector[T, list[T], list[T]]

Build a collector that stops after retaining the first count items.

A zero count completes before pulling from the source.

Parameters:

Name	Type	Description	Default
`count`	`int`	Non-negative maximum number of items to retain.	required

Returns:

Type	Description
`Collector[T, list[T], list[T]]`	An early-stopping collector returning an encounter-ordered list.

Raises:

Type	Description
`ValueError`	If `count` is negative.

tail `staticmethod`

tail(count: int) -> Collector[T, deque[T], list[T]]

Build a bounded-state collector for the final count input items.

The source is fully consumed. Once the deque is full, each new item discards the oldest; a zero count retains nothing.

Parameters:

Name	Type	Description	Default
`count`	`int`	Non-negative maximum number of trailing items to retain.	required

Returns:

Type	Description
`Collector[T, deque[T], list[T]]`	A collector finishing its bounded deque as an encounter-ordered list.

Raises:

Type	Description
`ValueError`	If `count` is negative.

only `staticmethod`

only() -> Collector[T, list[T], T | None]

Build a collector that enforces zero-or-one input cardinality.

Empty input returns None, one item is returned directly, and a second item completes collection early so finishing can raise without pulling a third.

Returns:

Type	Description
`Collector[T, list[T], T \| None]`	A collector retaining no more than two items.

Raises:

Type	Description
`ValueError`	If the input contains more than one item.

to_dict `staticmethod`

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

Build a dictionary collector with an explicit duplicate-key policy.

On duplicates, "error" raises before selecting the duplicate value, "first" preserves the existing entry without selecting another value, and "last" replaces it. Selector errors and unhashable keys propagate.

Parameters:

Name	Type	Description	Default
`key`	`Selector`	Selector deriving each dictionary key.	required
`value`	`Selector`	Selector deriving each dictionary value.	required
`on_duplicate`	`Literal['error', 'first', 'last']`	One of `"error"`, `"first"`, or `"last"`.	`'error'`

Returns:

Type	Description
`Collector[T, dict[K, V], dict[K, V]]`	A collector preserving key insertion order in its result dictionary.

Raises:

Type	Description
`ValueError`	If `on_duplicate` is not a supported policy.
`DuplicateKeyError`	If a repeated key is encountered under `"error"`.

to_columns `staticmethod`

to_columns() -> Collector[
    Mapping[str, Any], _ColumnsState, dict[str, list[Any]]
]

Build a collector that transposes variably shaped mapping rows into columns.

Columns retain field encounter order. Fields introduced by later rows are backfilled with None, and fields absent from later rows append None, so every column has one entry per input row.

Returns:

Type	Description
`Collector[Mapping[str, Any], _ColumnsState, dict[str, list[Any]]]`	A collector returning a dictionary of equally sized column lists.

Aggregators

agg creates named, single-pass aggregations:

Call	Result
`agg.count()`	Number of input items
`agg.count_where(predicate)`	Number of matching items
`agg.any(predicate)` / `agg.all(predicate)`	Boolean checks
`agg.sum(selector)` / `agg.mean(selector)`	Sum or arithmetic mean
`agg.variance(selector, ddof=1)`	Variance
`agg.std(selector, ddof=1)`	Standard deviation
`agg.count_distinct(selector)`	Number of distinct values
`agg.min(selector)` / `agg.max(selector)`	Smallest or largest value
`agg.first(selector)` / `agg.last(selector)`	Boundary values
`agg.collect(selector, into=list)`	Values collected with a constructor

Omit selector to aggregate the items themselves. A string selector reads a record field; expression and callable selectors are also accepted.

fpstreams.Aggregator `dataclass`

Bases: Collector[Any, Any, Any]

A collector accepted by named aggregate terminals and native fusion.

Collecting and aggregation

Collector

fpstreams.Collector dataclass

Built-in collectors

fpstreams.Collectors

to_list staticmethod

to_set staticmethod

to_tuple staticmethod

joining staticmethod

grouping_by staticmethod

partitioning_by staticmethod

mapping staticmethod

filtering staticmethod

flat_mapping staticmethod

collecting_and_then staticmethod

teeing staticmethod

counting staticmethod

summing staticmethod

averaging staticmethod

summarizing staticmethod

first staticmethod

last staticmethod

head staticmethod

tail staticmethod

only staticmethod

to_dict staticmethod

to_columns staticmethod

Aggregators

fpstreams.Aggregator dataclass

fpstreams.Collector `dataclass`

to_list `staticmethod`

to_set `staticmethod`

to_tuple `staticmethod`

joining `staticmethod`

grouping_by `staticmethod`

partitioning_by `staticmethod`

mapping `staticmethod`

filtering `staticmethod`

flat_mapping `staticmethod`

collecting_and_then `staticmethod`

teeing `staticmethod`

counting `staticmethod`

summing `staticmethod`

averaging `staticmethod`

summarizing `staticmethod`

first `staticmethod`

last `staticmethod`

head `staticmethod`

tail `staticmethod`

only `staticmethod`

to_dict `staticmethod`

to_columns `staticmethod`

fpstreams.Aggregator `dataclass`