Flow

Flow[T] is the primary synchronous lazy pipeline for both ordinary values and records. Transformations return a new pipeline; terminal methods execute its plan.

Use flow(source) for an existing source and flow.defer(factory) when every execution must create a fresh iterable. Besides ordinary iterables, flow(source) retains supported PyArrow, pandas, Polars, Arrow C stream, and dataframe-interchange inputs.

chunk(size, strict=True) (2.5.0) rejects an incomplete final batch when the flow reaches it. The default continues to emit a shorter final batch. batch is an alias; AsyncFlow.chunk provides the same option. In strict mode, the short tail has already been read when ValueError is raised; it is not emitted. To inspect a short tail yourself, use the default mode and check the tuple length. Stopping before that tail does not validate it.

In the 2.5.0 source, map_parallel and parallel normalize non-None workers and buffer values with the integer __index__ protocol when configuring the flow, before opening its source. Values must be positive. Floating-point values, including NaN and infinity, raise TypeError.

Consumption, batching, and cleanup

In the 2.5.0 Python uniqueness path, distinct and distinct_by also compare hashable keys with previously seen unhashable keys, and vice versa. The first accepted equal key wins. Mixed categories can invoke additional equality callbacks and propagate their errors. State retains distinct keys; with H hashable and U unhashable keys, cross-category comparisons may take O(H × U) time and state uses O(H + U) references. Existing same-category comparison rules remain unchanged. For a new hashable key, each previously accepted unhashable key is compared as seen == key; for a new unhashable key, accepted hashable keys are compared as key == seen in encounter order.

group_runs emits each contiguous group as a complete tuple. To detect a group boundary, it reads the first item of the next group and computes its key before emitting the previous group. If that key computation fails, the previous group has not yet been emitted. It holds the whole current group in memory; the tuple contains the original item references, without deep copies. An infinite run with no key change cannot produce a completed group.

In the 2.5.0 source, concat opens each additional source only when it reaches it and closes an opened iterator on completion or early exit. Passing the same one-shot iterator as the input and as an additional source does not give the additional source a second ownership of that iterator.

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

zip pulls the left input before the right input. If the right input ends first, a left item can already have been consumed without appearing in a pair. This also matters when the inputs have side effects. strict=True checks for unequal lengths during consumption, so a downstream take cannot validate an unconsumed tail. zip_longest uses None for padding by default, which is indistinguishable from an input None. Pass a unique sentinel when the distinction matters. A mutable fillvalue is reused, not copied.

intersperse reads the next input item before emitting its separator. Consequently, intersperse(...).take(2) can consume two source items even though only the first item and a separator are returned. A successful first or find only checks the consumed prefix; consume the entire source if all items and source errors must be checked.

batch_by_size checks the next item's size before emitting a pending batch, including when max_count has already been reached. An oversized next item in strict mode or an error from the source or size function can therefore stop the flow before that pending batch is emitted. Normal exhaustion emits the remaining nonempty batch. The operation does not copy the items in its tuples.

With map_parallel(..., ordered=False), results arrive in completion order; following it with take(n) does not select the first n inputs. Work already submitted can have started even when its result is never returned.

Terminal operations own their iteration and close it when they stop. When iterating directly, keep and explicitly close the iterator if you stop early; break alone does not close a retained iterator. In the Python path, direct iteration of take(1) can yield its last item while the source remains open; cleanup occurs when iteration resumes or the iterator is explicitly closed. For an iterator with a close method, contextlib.closing can express this ownership:

from contextlib import closing
from fpstreams import flow

with closing(iter(flow(range(10)))) as values:
    for value in values:
        if value == 2:
            break

For a direct flow.defer(factory) source, take(0).to_list() opens and closes the factory's iterator without pulling an item, while collect(Collectors.head(0)) does not call the factory. Factory errors in the first case still propagate. Custom iterators need not implement close.

Flattening mapped iterables

flat_map() expands one iterable level in order. A string is iterable and therefore yields its characters; a dictionary yields its keys. Wrap either in another iterable to keep it as one value.

In the synchronous Python path, a mapped iterator is delegated with yield from. It is not explicitly closed when it is exhausted normally or when its own __next__ raises. Closing an active delegated iterator after downstream early termination calls its close() if available. A generator may also run its own cleanup as it finishes; these details do not imply that every iterable provides a close hook.

Scan state and Gatherer completion

scan(initial, function) emits one state per input and emits nothing for an empty input. It does not copy the initial state or results. If a callback mutates and returns one object, previously emitted references can observe later mutations; the supplied seed is also reused on another evaluation. Return fresh values for snapshots, and create a fresh pipeline and seed with defer when each execution needs independent state:

from fpstreams import flow

def add_key(counts, key):
    return {**counts, key: counts.get(key, 0) + 1}

counts = flow.defer(lambda: flow(["a", "b", "a"]).scan({}, add_key))
expected = [{"a": 1}, {"a": 1, "b": 1}, {"a": 2, "b": 1}]
assert counts.to_list() == expected
assert counts.to_list() == expected

scan_right buffers the full source and computes the right fold before its first output; it emits nothing for an empty source. With max_items, it must read the next item to detect an exceeded limit, so the limit bounds buffered items rather than all source reads.

For the synchronous Gatherer, the finisher runs on normal exhaustion and when the gatherer stops input. Downstream early close can also run it with a rejecting output channel. It must not infer that the source was exhausted from the number of items seen. Source or integration failures do not turn the finisher into a finally hook; manage separately owned resources explicitly. In a composed gatherer, an initializer or finisher failure can prevent later hooks from running.

Reduce values and check their types

reduce(function, initial=...) combines values from left to right. Without an initial value, the first element is the accumulator; an empty flow raises EmptyFlowError. With an initial value, an empty flow returns that exact object. The callback must accept its own previous result as the next accumulator.

The current declaration accepts Callable[[Any, T], Any] and returns Any, with or without an initial value. Runtime results can be more specific, but neither an ordinary lambda nor a named, annotated callback refines that public return type:

from typing import Any, assert_type
from fpstreams import flow

assert_type(flow([1, 2]).reduce(lambda a, b: a + b), Any)  # runtime: 3
assert_type(flow([1, 2]).reduce(lambda a, b: a + str(b), ""), Any)  # runtime: "12"

Any can also let an invalid callback pass static checking. For example, def step(a: int, b: int) -> str: return str(a + b) works for [1, 2] without an initial value, but fails with TypeError for [1, 2, 3]: its next accumulator is a string, which the callback cannot add to an integer.

For an explicit accumulator contract, use the existing lazy fold operation and annotate its result. This gives ordinary lambdas concrete parameter types in both mypy and Pyright. The initializer creates state, and .first() returns the one emitted result:

from fpstreams import Flow, flow


def initial_text() -> str:
    return ""


query: Flow[str] = flow([1, 2]).fold(
    initial_text, lambda state, item: state + str(item)
)
assert query.first() == "12"
assert Flow[int]([]).fold(initial_text, lambda state, item: state).first() == ""

This is a migration from reduce, whose return declaration remains Any. fold requires an initializer and calls the step for every input item, including the first. It emits the initializer's exact result on empty input and never awaits state values. To migrate an unseeded reduction, choose an explicit seed that suits the operation; a singleton now also runs the step. The initializer runs once per evaluation, so use a factory when each evaluation needs fresh mutable state. A callback or seed containing Any can bypass the declared contract. These annotations add no runtime state checks and do not turn list[Any] into list[int]. For untrusted dynamic results, keep the runtime check described below.

For a concrete result, receive it as object and check it before use:

from fpstreams import flow

result: object = flow([1, 2]).reduce(lambda a, b: a + b)
if not isinstance(result, int):
    raise TypeError("expected an integer accumulator")
assert result == 3  # narrowed to int

When several calls share a specific accumulator contract, a local boundary can check the callback signature and the returned value:

from collections.abc import Callable
from fpstreams import Flow, flow


def text_step(accumulator: str, item: int) -> str:
    return accumulator + str(item)


def reduce_text(values: Flow[int], step: Callable[[str, int], str], initial: str) -> str:
    result: object = values.reduce(step, initial)
    if not isinstance(result, str):
        raise TypeError("text reducer returned a non-string")
    return result


callback: Callable[[str, int], str] = text_step
assert reduce_text(flow([1, 2]), callback, "") == "12"

This is a caller-owned string example, not an additional fpstreams API or a promise of arbitrary generic safety. A callback returning Any can still bypass static signature checks; the explicit result check above catches a non-string result if the reduction returns. It does not validate every intermediate value. A nullable accumulator must accept None; nested Any, such as list[Any], still carries unknown element types. Neither should be relabelled as a more precise type without evidence.

In a source checkout, tests/typing/reduce_examples.py contains the runnable examples, nullable and nested-Any cases, identity checks, and a runtime witness for the invalid callback. tests/typing/reduce_invalid.py contains marked expected static errors and must not be executed. From the checkout root, the unified uv run python scripts/check_typing.py command checks both example pairs with mypy and Pyright, one invocation per checker. See the repository's CONTRIBUTING "Run the checks" section for setup and target selection. No optional data backend is needed to run the reduce examples.

Statistical pipelines (2.5.0)

Use collect(Collectors.statistics(...)) for one summary over the whole input, or rolling_statistics(size, selector=..., ddof=0) for one result per input item:

from fpstreams import Collectors, flow

assert flow([2, None, 4, 6]).collect(Collectors.statistics())["mean"] == 4.0
assert flow([2, None, 4, 6]).rolling_statistics(2).to_list() == [
    (1, 2.0, 0.0), (1, 2.0, 0.0), (1, 4.0, 0.0), (2, 5.0, 1.0)
]

Rolling results are (count, mean, variance) tuples, including warmup windows. Selected None occupies a window position and contributes no observation. Size and ddof use the integer index protocol: invalid types raise TypeError, and a nonpositive size or negative ddof raises ValueError at construction. Non-real inputs raise TypeError during consumption; nonfinite rolling inputs raise ValueError. Each evaluation starts with fresh window state.

See shared and approximate statistics for Collector input policies and optional sketches, and the rolling recipe for state bounds and rebuilding costs. These additions require the source version.

State-driven generation and mapping (2.4.0)

unfold builds a source from state, while map_accum transforms an existing source with state. They use different tuple orders:

Operation	Callback result	Emitted value
flow.unfold(seed, function)	(value, next_state) or None to stop	value
query.map_accum(initial, function)	(next_state, output)	output

from fpstreams import flow

countdown = flow.unfold(3, lambda n: (n, n - 1) if n else None)
assert countdown.to_list() == [3, 2, 1]
assert countdown.take(1).to_list() == [3]

The unfold seed is not emitted by itself. Returning (None, next_state) emits an actual None; only returning None ends generation. The callback controls termination, so bound an infinite transition with take or another stopping operation. Construction and take(0) do not run the callback.

from fpstreams import flow


def running_total(total, value):
    next_total = total + value
    return next_total, {"value": value, "total": next_total}


query = flow([2, 3, 5]).map_accum(0, running_total)
assert query.to_list() == [
    {"value": 2, "total": 2},
    {"value": 3, "total": 5},
    {"value": 5, "total": 10},
]

map_accum calls the function once per consumed item in encounter order. It emits nothing for an empty source and does not emit its final state. Each evaluation starts from the supplied initial state; mutable seed objects are reused by reference, following scan. Use flow.defer to allocate fresh mutable state for each evaluation. The operation keeps the source's replay and cleanup rules and runs its state transitions sequentially even with parallel map defaults.

Reuse sliding-window aggregates (2.4.0)

window_reduce(size, combine, step=1) combines each count-based window without building a tuple and folding every overlapping value again:

from operator import add
from fpstreams import flow

assert flow([2, 3, 5, 7]).window_reduce(3, add).to_list() == [10, 15]
assert flow([2, 3]).window_reduce(3, add).to_list() == [5]

combine must be associative and must not modify either argument. Source values and returned aggregates must remain unchanged during iteration because later windows may reuse them. Encounter order is preserved, so a noncommutative associative function is supported. Floating-point addition is not associative; use window(...).map(...) when a left fold or arbitrary callback behavior matters. There is no automatic conversion of existing window pipelines.

Empty input emits nothing. Initially short input emits one partial result; after a full window, incomplete trailing windows are omitted. step may exceed size; values in the gaps are consumed without being combined. Overlapping windows can group calls differently from a left fold. A combine error can arise while building a trailing window that would never be emitted. Stopping downstream closes the owned iterator chain without pulling another item.

Overlapping windows use two aggregate stacks, following the storage insight in Two-Stacks Lite. They cache at most size + 1 entries. When step >= size, a running left fold replaces the stacks and keeps no raw window buffer. Both paths preserve input order and fold incomplete tails, even when those tails produce no output. Cached entries do not bound bytes: an aggregate can grow with its input. Combine calls are amortized constant per input item; transferring a stack can take linear work in size. This does not provide DABA's worst-case constant operation bound. Small windows and windows without overlap may cost more than an ordinary fold; compare against window(...).map(...) for those workloads.

Reduce adjacent runs with a Collector (2.4.0)

collapse(predicate, collector) feeds each nonempty adjacent run into fresh Collector state. The predicate sees neighboring input items, including when it is not transitive:

from fpstreams import Collectors, flow

query = flow([1, 2, 4, 9, 11, 1]).collapse(
    lambda left, right: 0 <= right - left <= 2,
    Collectors.counting(),
)
assert query.to_list() == [3, 2, 1]

Each run has its own initializer, steps, completion state, and finisher. Empty input emits nothing and initializes no collector. A completed collector stops receiving steps, while adjacent predicates keep running until the run boundary. For example, Collectors.head(1) keeps only the first item of each run; it cannot finish the whole query after seeing the first item.

A boundary consumes one item of the next run before emitting the current result. The next collector is initialized when consumption resumes, so take(1) does not initialize an unused next run. The last run is finished after normal source exhaustion. Source errors and early termination do not flush a pending run. An infinite run still needs a boundary or source exhaustion to emit.

The operation keeps collector state and adjacent input references without a raw run buffer. Collectors.counting() keeps one running integer; Collectors.to_list() retains every run member. A growing collector state and the final to_list() result can still grow with input size. Collector hooks run while reading input, so callback errors can appear earlier than with group_runs(...).map(collector). This is an explicit overload, not an automatic rewrite of buffered pipelines.

The adjacent predicate and the state program add work as input arrives. When grouping by equal keys and the groups fit in memory, compare this overload with group_runs(...).map(collector). Buffered execution can be faster.

Hooks follow the synchronous Collector protocol, including done. A finisher returning an awaitable emits that object as data. Collector combine and native metadata are unused by this sequential reduction. The binary merger form of collapse(predicate, merger) remains available.

Record operations and the Rows view

Flow.rows() creates a lazy relational view without inspecting any item. The view shares the same plan and source ownership, so it does not make a one-shot input reusable.

Record operations whose names do not conflict with Flow semantics can be called directly. select(), with_columns(), rename(), cast(), fill_nulls(), drop_nulls(), explode(), unnest(), unpivot(), and pivot() return Rows; group_by() returns GroupedRows.

Four relation-building names retain their established Flow meanings. Enter the Rows view first for these relational forms:

Flow call	Flow meaning	Explicit Rows call	Rows meaning
drop(count)	Skip leading items	rows().drop(*columns)	Remove record fields
join(separator)	Join string representations	rows().join(other, ...)	Relational join
aggregate(...)	Execute and return a dictionary	rows().aggregate(...)	Build a lazy one-row relation
where(predicate)	Alias of filter	rows().where(predicate, **equalities)	Filter records, including field equalities

This table covers relation-building operations, not every shared method name. Flow keeps its own output signatures too. Enter .rows() before to_csv(), to_pandas(), or to_df() when you need Rows-specific writer or materializer options.

Rows.map() and Rows.flat_map() return an ordinary Flow because their output may have any shape. If those functions produce records, a later nonconflicting record method can enter Rows again.

Tabular source routing

When the corresponding package is already loaded, flow(source) recognizes concrete PyArrow Table, RecordBatch, and RecordBatchReader objects, pandas DataFrames, and Polars DataFrame or LazyFrame objects. It does not import those packages merely to probe an arbitrary object. A custom object with __arrow_c_stream__ or __dataframe__ is routed through the matching adapter; Arrow wins when both protocols are present. Pandas conversion emits data columns only, not the dataframe index.

This routing does not sample ordinary iterable contents. Built-in containers, generators, NumPy arrays, and plain two-dimensional lists are not classified by their contents. Record methods still work when their items support the requested selectors. File paths are not guessed as CSV or Parquet inputs; use an explicit factory.

Conversion timing follows the selected protocol:

• an ordinary generator is not opened during construction;
• generic dataframe conversion and Polars LazyFrame collection are deferred until execution;
• a custom Arrow C stream is imported once during construction and is one-shot;
• a PyArrow RecordBatchReader is one-shot, while retained tables and record batches are reiterable.

Explaining terminal execution

explain() defaults to ordinary iteration. Pass a terminal name when you want to inspect to_list(), count(), sum(), statistics, aggregation, or a short-circuiting terminal. The same planner is used by the explanation and the terminal itself.

from fpstreams import flow

explanation = flow([1, 2, 3]).explain(terminal="count").to_dict()

assert explanation["selected_engine"] == "python"
assert explanation["complexity"] == "O(1)"
assert explanation["semantics"]["output"]["cardinality"] == {
    "kind": "exact",
    "value": 3,
}
assert explanation["diagnostics"] == []
assert explanation["arrow_prefix"] is None
assert explanation["boundaries"] == []

An exact cardinality describes the number of outputs on normal, complete consumption under the source facts available when explain() runs. It does not guarantee that callbacks succeed or finish, and a referenced mutable source can change before execution. upper_bound limits the outputs under the same source facts; it does not guarantee normal completion. When a strict zip or chunk is known to fail on a length mismatch, the planner reports a conservative upper bound rather than an exact count. A downstream take may stop before that failure is reached.

An Arrow-capable plan reports its retained prefix and any guarded transition to Python rows in arrow_prefix and boundaries. Relational plans additionally report their selected tree and strategy in relations.

run_with_report() returns the terminal value, its recorded route, and query-owned resource counts in one execution. The report covers the outer plan and some direct paths; internal kernels and fallbacks are not all recorded.

An identity list or tuple remains in Python under auto when a terminal would otherwise scan and copy it. An identity range can still use native numeric reduction. Exact-size count() is O(1) only for an unchanged, safely reiterable source; operations and one-shot inputs are consumed normally.

CSV safety

Flow to_csv(..., spreadsheet_safe=False) writes arbitrary scalar, sequence, or mapping items and accepts an optional header. Set spreadsheet_safe=True when untrusted strings will be opened in Excel, Sheets, or similar software. Suspect formula prefixes are neutralized with a leading single quote. The method writes the file and returns None.

After entering Rows, to_csv() is the record writer instead and exposes fieldnames, include_header, and extrasaction options.

Creating a flow

flow(range(...)) and aflow(range(...)) accept ranges longer than sys.maxsize. Bounded reads remain lazy, and an identity count() returns the exact cardinality.

Call	Behavior
flow(source)	Wrap an ordinary iterable, reuse a Flow/Rows plan, or route a supported tabular source
flow.defer(factory)	Call the factory for each execution
flow.from_arrow(source)	Adapt an Arrow table, batch, reader, or C stream provider
flow.from_columns(columns)	Build and retain an Arrow table from equal-length named columns
flow.from_numpy(array, columns=...)	Adapt a one-dimensional array to scalars or a two-dimensional array to named records
flow.from_dataframe(frame)	Use the dataframe interchange protocol; from_pandas is an alias
flow.from_polars(frame)	Retain a Polars DataFrame or LazyFrame
flow.scan_csv(path)	Scan typed Arrow CSV batches; reader column selection requires explicit conversion options
flow.from_parquet(source)	Scan Parquet with optional projection and filtering
flow.empty()	Create a flow with no items
flow.of_nullable(value)	Emit one value, or nothing when it is None
flow.iterate(seed, function)	Repeatedly derive the next value from the previous one
flow.generate(supplier)	Call a supplier for every emitted value
flow.concat(*sources)	Read several sources in order

Methods

fpstreams.Flow

Bases: FlowTerminalsMixin[T], Generic[T]

A synchronous pipeline that opens its source only when iterated or consumed.

run_with_report

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

Run one eager terminal normally and return its value with a read-only report.

terminal names an existing eager method such as "to_list" or "sum". Lazy transformations and iteration are deliberately excluded.

to_list

to_list() -> list[T]

Execute the pipeline and collect its items in a list.

Returns:

Type	Description
list[T]	All emitted items in encounter order.

to_tuple

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

Execute the pipeline and collect its items in a tuple.

Returns:

Type	Description
tuple[T, ...]	All emitted items in encounter order as a tuple.

to_set

to_set() -> set[T]

Execute the pipeline and collect distinct hashable items.

Returns:

Type	Description
set[T]	The distinct emitted items; every item must be hashable.

to_pandas

to_pandas(columns: Iterable[str] | None = None) -> Any

Execute the pipeline and build a pandas DataFrame.

Parameters:

Name	Type	Description	Default
columns	Iterable[str] | None	Optional column labels passed to pandas.DataFrame.	None

Returns:

Type	Description
Any	A pandas DataFrame containing the emitted items.

to_numpy

to_numpy(dtype: Any = None) -> Any

Execute the pipeline and build a NumPy array.

Parameters:

Name	Type	Description	Default
dtype	Any	The optional NumPy data type used for the resulting array.	None

Returns:

Type	Description
Any	A NumPy array containing the emitted items.

to_csv

to_csv(
    path: str | PathLike[str],
    *,
    header: Iterable[str] | None = None,
    encoding: str = "utf-8",
    spreadsheet_safe: bool = False,
    atomic: bool = False,
    if_exists: Literal["replace", "error"] = "replace",
) -> None

Execute the pipeline and stream its items to a CSV file.

Parameters:

Name	Type	Description	Default
path	str | PathLike[str]	Destination file, opened in text write mode.	required
header	Iterable[str] | None	Optional header row. For mapping items, these names also select and order cells.	None
encoding	str	Encoding used when opening the destination.	'utf-8'
spreadsheet_safe	bool	Prefix formula-like string cells, including headers, so spreadsheet software treats them as text. Mapping keys used for lookup remain unchanged.	False
atomic	bool	Publish after the writer and source close successfully. Writing or source-cleanup failures preserve the old target.	False
if_exists	Literal['replace', 'error']	With atomic=True, replace the target or fail if it already exists.	'replace'

to_json

to_json(
    path: str | PathLike[str],
    *,
    encoding: str = "utf-8",
    ensure_ascii: bool = False,
    default: Callable[[Any], Any] | None = None,
    atomic: bool = False,
    if_exists: Literal["replace", "error"] = "replace",
) -> None

Execute the pipeline and stream its items to one JSON array.

Parameters:

Name	Type	Description	Default
path	str | PathLike[str]	Destination file, replaced with one streamed JSON array.	required
encoding	str	Encoding used when opening the destination.	'utf-8'
ensure_ascii	bool	Whether JSON output escapes non-ASCII characters.	False
default	Callable[[Any], Any] | None	Optional serializer called for objects the JSON encoder cannot handle.	None
atomic	bool	Publish after the writer and source close successfully. Writing or source-cleanup failures preserve the old target.	False
if_exists	Literal['replace', 'error']	With atomic=True, replace the target or fail if it already exists.	'replace'

to_jsonl

to_jsonl(
    path: str | PathLike[str],
    *,
    encoding: str = "utf-8",
    ensure_ascii: bool = False,
    default: Callable[[Any], Any] | None = None,
    atomic: bool = False,
    if_exists: Literal["replace", "error"] = "replace",
) -> None

Stream each item as one JSON value followed by a newline.

Values may be scalars, sequences, or records. An empty flow writes an empty file. This terminal consumes the source once without collecting the output in memory.

Parameters:

Name	Type	Description	Default
path	str | PathLike[str]	Destination JSON Lines file.	required
encoding	str	Encoding used when opening the destination.	'utf-8'
ensure_ascii	bool	Whether JSON output escapes non-ASCII characters.	False
default	Callable[[Any], Any] | None	Optional serializer called for objects the JSON encoder cannot handle.	None
atomic	bool	Publish after the writer and source close successfully. Writing or source-cleanup failures preserve the old target.	False
if_exists	Literal['replace', 'error']	With atomic=True, replace the target or fail if it already exists.	'replace'

describe

describe() -> dict[str, int | float]

Return count and one-pass summary statistics for numeric items.

Returns:

Type	Description
dict[str, int | float]	A dictionary containing count, or {} for an empty flow. Real-valued items add sum, min, max, and mean. Mixed input also adds numeric_count; two or more numeric items add sample standard deviation as std.

aggregate

aggregate(
    **aggregations: Collector[Any, Any, Any],
) -> dict[str, Any]

Compute several named aggregations while traversing the flow once.

Collectors and agg factory results share the same source traversal.

Parameters:

Name	Type	Description	Default
**aggregations	Collector[Any, Any, Any]	Result names mapped to collectors updated in one traversal.	{}

Returns:

Type	Description
dict[str, Any]	Finished aggregation values keyed by the supplied argument names.

collect

collect(
    collector: Collector[T, S, C],
    /,
    **collectors: Collector[T, Any, Any],
) -> C

collect(
    collector: Callable[[Iterable[T]], C],
    /,
    **collectors: Collector[T, Any, Any],
) -> C

collect(
    collector: None = None,
    /,
    **collectors: Collector[T, Any, Any],
) -> dict[str, Any]

collect(
    collector: Callable[[Iterable[T]], C]
    | Collector[T, Any, C]
    | None = None,
    /,
    **collectors: Collector[T, Any, Any],
) -> C | dict[str, Any]

Reduce the pipeline with one Collector or named Collectors.

One collector returns its finished value. Named collectors share one source traversal and return a dictionary.

Parameters:

Name	Type	Description	Default
collector	Callable[[Iterable[T]], C] | Collector[T, Any, C] | None	One Collector or callable invoked with this flow to produce the result.	None
**collectors	Collector[T, Any, Any]	Named streaming collectors updated together in one traversal.	{}

Returns:

Type	Description
C | dict[str, Any]	The collector result, or a dictionary of results for named collectors.

join

join(separator: str = '') -> str

Convert items to strings and join them with separator.

This is a string terminal operation; it consumes the flow and does not perform a relational join.

Parameters:

Name	Type	Description	Default
separator	str	String placed between consecutive str(item) values.	''

Returns:

Type	Description
str	One string containing every item separated by separator.

for_each

for_each(action: Callable[[T], Any]) -> None

Execute action once for every item.

Parameters:

Name	Type	Description	Default
action	Callable[[T], Any]	Called once for each emitted item; its return value is ignored.	required

partition

partition(
    predicate: Callable[[T], bool],
) -> tuple[list[T], list[T]]

Collect matching and non-matching items in separate lists.

Parameters:

Name	Type	Description	Default
predicate	Callable[[T], bool]	Called once per item; truthy items enter the first returned list.	required

Returns:

Type	Description
tuple[list[T], list[T]]	(matches, misses), preserving encounter order within both lists.

partition_results

partition_results() -> tuple[list[Any], list[Exception]]

Separate Result values into successes and failures.

Returns:

Name	Type	Description
success_values	list[Any]	Unwrapped Ok values in encounter order.
exceptions	list[Exception]	Exceptions stored by Err, in encounter order.

Raises:

Type	Description
TypeError	If any emitted item is neither Ok nor Err.

to_async

to_async() -> fpstreams.AsyncFlow[T]

View this synchronous pipeline as an AsyncFlow.

Returns:

Type	Description
AsyncFlow[T]	An AsyncFlow that yields the same items.

first

first() -> T

first(default: R) -> T | R

first(default: Any = _MISSING) -> T | Any

Return the first item without consuming an unnecessary tail.

Parameters:

Name	Type	Description	Default
default	Any	Returned only when the flow is empty.	_MISSING

Returns:

Type	Description
T | Any	The first item, or default when the flow is empty.

Raises:

Type	Description
EmptyFlowError	If the flow is empty and no default is supplied.

last

last() -> T

last(default: R) -> T | R

last(default: Any = _MISSING) -> T | Any

Return the last item, or default when the flow is empty.

Parameters:

Name	Type	Description	Default
default	Any	Returned only when the flow is empty.	_MISSING

Returns:

Type	Description
T | Any	The last item, or default when the flow is empty.

Raises:

Type	Description
EmptyFlowError	If the flow is empty and no default is supplied.

find

find(predicate: Callable[[T], Any]) -> T

find(predicate: Callable[[T], Any], default: R) -> T | R

find(
    predicate: Callable[[T], Any], default: Any = _MISSING
) -> T | Any

Return the first matching item, a default, or raise EmptyFlowError.

Parameters:

Name	Type	Description	Default
predicate	Callable[[T], Any]	Called in order until its first truthy result.	required
default	Any	Returned when no predicate result is truthy.	_MISSING

Returns:

Type	Description
T | Any	The first matching item, or default when no item matches.

Raises:

Type	Description
EmptyFlowError	If no item matches and no default is supplied.

find_index

find_index(predicate: Callable[[T], Any]) -> int | None

Return the index of the first matching item, or None.

Parameters:

Name	Type	Description	Default
predicate	Callable[[T], Any]	Called with each item in order until its first truthy result.	required

Returns:

Type	Description
int | None	The zero-based position of the first truthy predicate result, or None.

index_of

index_of(value: T) -> int | None

Return the index of the first equal value, or None.

Parameters:

Name	Type	Description	Default
value	T	Target compared to each source item with equality.	required

Returns:

Type	Description
int | None	The zero-based position of the first item equal to value, or None.

nth

nth(index: int) -> T

nth(index: int, default: R) -> T | R

nth(index: int, default: Any = _MISSING) -> T | Any

Return the item at a positive or negative index.

Parameters:

Name	Type	Description	Default
index	int	Zero-based position; negative values count backward from the end.	required
default	Any	Returned when index is outside the flow.	_MISSING

Returns:

Type	Description
T | Any	The selected item, or default when the index is out of range.

Raises:

Type	Description
EmptyFlowError	If the index is out of range and no default is supplied.

count

count() -> int

Count all items produced by the pipeline.

Returns:

Type	Description
int	The total number of emitted items.

sum

sum(start: Any = 0) -> Any

Add all items, starting with start.

Parameters:

Name	Type	Description	Default
start	Any	Value added before all emitted items, matching Python's built-in sum.	0

Returns:

Type	Description
Any	The total of start and all emitted items.

mean

mean() -> float | None

Return the arithmetic mean, or None for an empty flow.

Returns:

Type	Description
float | None	The compensated floating-point mean, or None when no items are emitted.

Raises:

Type	Description
TypeError	If an emitted item is not a real number.

variance

variance(*, ddof: int = 1) -> float | None

Return the variance, or None when too few values are available.

Parameters:

Name	Type	Description	Default
ddof	int	Non-negative adjustment in the divisor count - ddof.	1

Returns:

Type	Description
float | None	The floating-point variance, or None when count <= ddof.

Raises:

Type	Description
TypeError	If an emitted item is not a real number.
ValueError	If ddof is negative.

std

std(*, ddof: int = 1) -> float | None

Return the standard deviation, or None when too few values are available.

Parameters:

Name	Type	Description	Default
ddof	int	Non-negative adjustment in the variance divisor count - ddof.	1

Returns:

Type	Description
float | None	The square root of the variance, or None when count <= ddof.

Raises:

Type	Description
TypeError	If an emitted item is not a real number.
ValueError	If ddof is negative.

min

min(*, key: Callable[[T], Any] | None = None) -> T

Return the smallest item and raise on an empty flow.

Parameters:

Name	Type	Description	Default
key	Callable[[T], Any] | None	Optional callable whose result is compared instead of the item.	None

Returns:

Type	Description
T	The smallest item according to key.

max

max(*, key: Callable[[T], Any] | None = None) -> T

Return the largest item and raise on an empty flow.

Parameters:

Name	Type	Description	Default
key	Callable[[T], Any] | None	Optional callable whose result is compared instead of the item.	None

Returns:

Type	Description
T	The largest item according to key.

top

top(count: int, *, key: Selector | None = None) -> list[T]

Return up to count largest items without sorting the entire result.

Parameters:

Name	Type	Description	Default
count	int	Maximum number of items to return.	required
key	Selector | None	Optional callable, field name, index, path, or expression used for ranking.	None

Returns:

Type	Description
list[T]	Up to count items ordered from largest to smallest selected value.

bottom

bottom(
    count: int, *, key: Selector | None = None
) -> list[T]

Return up to count smallest items without sorting the entire result.

Parameters:

Name	Type	Description	Default
count	int	Maximum number of items to return.	required
key	Selector | None	Optional callable, field name, index, path, or expression used for ranking.	None

Returns:

Type	Description
list[T]	Up to count items ordered from smallest to largest selected value.

minmax

minmax(*, key: Selector | None = None) -> tuple[T, T]

Return the smallest and largest items in one traversal.

Parameters:

Name	Type	Description	Default
key	Selector | None	Optional callable, field name, index, path, or expression used for comparison.	None

Returns:

Type	Description
tuple[T, T]	(minimum_item, maximum_item) according to the selected values.

Raises:

Type	Description
EmptyFlowError	If the flow emits no items.

any

any(predicate: Callable[[T], bool] = bool) -> bool

Return whether at least one item satisfies predicate.

Parameters:

Name	Type	Description	Default
predicate	Callable[[T], bool]	Tested in order until one result is truthy; defaults to bool.	bool

Returns:

Type	Description
bool	True when any item satisfies predicate; False for an empty flow.

all

all(predicate: Callable[[T], bool] = bool) -> bool

Return whether every item satisfies predicate.

Parameters:

Name	Type	Description	Default
predicate	Callable[[T], bool]	Tested in order until one result is falsey; defaults to bool.	bool

Returns:

Type	Description
bool	True when every item satisfies predicate, including for an empty flow.

none

none(predicate: Callable[[T], bool] = bool) -> bool

Return whether no item satisfies predicate.

Parameters:

Name	Type	Description	Default
predicate	Callable[[T], bool]	Tested in order until one result is truthy; defaults to bool.	bool

Returns:

Type	Description
bool	True only when no item satisfies predicate, including for an empty flow.

reduce

reduce(
    function: Callable[[Any, T], Any],
    initial: Any = _MISSING,
) -> Any

Combine items from left to right with an optional initial value.

Parameters:

Name	Type	Description	Default
function	Callable[[Any, T], Any]	Called as function(accumulator, item) from left to right.	required
initial	Any	Starting accumulator; when omitted, the first item becomes the accumulator.	_MISSING

Returns:

Type	Description
Any	The final left-to-right accumulator.

Raises:

Type	Description
EmptyFlowError	If the flow is empty and initial is omitted.

reduce_right

reduce_right(
    function: Callable[[T, Any], Any],
    initial: Any = _MISSING,
    *,
    max_items: int | None = None,
) -> Any

Combine buffered items from right to left.

Parameters:

Name	Type	Description	Default
function	Callable[[T, Any], Any]	Called as function(item, accumulator) from right to left.	required
initial	Any	Starting accumulator; when omitted, the last item becomes the accumulator.	_MISSING
max_items	int | None	Optional maximum number of source items that may be buffered.	None

Returns:

Type	Description
Any	The final right-to-left accumulator.

Raises:

Type	Description
BufferLimitError	If the source contains more than max_items items.
EmptyFlowError	If the flow is empty and initial is omitted.

reduce_by

reduce_by(
    key: Selector,
    function: Callable[[R, T], R],
    *,
    initializer: Callable[[], R],
) -> dict[Any, R]

Reduce items independently for each selected key.

Parameters:

Name	Type	Description	Default
key	Selector	Callable, field name, index, path, or expression selecting each group key.	required
function	Callable[[R, T], R]	Called as function(group_state, item) for items in that group.	required
initializer	Callable[[], R]	Called once when each distinct group is first encountered.	required

Returns:

Type	Description
dict[Any, R]	Final accumulator state for each hashable key, in first-key encounter order.

frequencies

frequencies(key: Selector | None = None) -> dict[Any, int]

Count occurrences of values or selected keys.

Parameters:

Name	Type	Description	Default
key	Selector | None	Optional callable, field name, index, path, or expression selecting the value to count; the item itself is counted when omitted.	None

Returns:

Type	Description
dict[Any, int]	Occurrence counts keyed by each hashable selected value.

of staticmethod

of(*items: R) -> Flow[R]

Create a flow from positional items.

Parameters:

Name	Type	Description	Default
*items	R	Positional values to emit in argument order.	()

Returns:

Type	Description
Flow[R]	A reusable flow that emits items in argument order.

from_iterable staticmethod

from_iterable(source: Iterable[R]) -> Flow[R]

Create a flow over an iterable source.

Parameters:

Name	Type	Description	Default
source	Iterable[R]	A synchronous iterable opened when the flow is consumed. Iterator instances remain one-shot; other iterables can be evaluated repeatedly.	required

Returns:

Type	Description
Flow[R]	A flow that emits each item from source in its iteration order.

empty staticmethod

empty() -> Flow[Any]

Create a flow that emits no items.

Returns:

Type	Description
Flow[Any]	A reusable flow that always completes without emitting an item.

of_nullable staticmethod

of_nullable(value: R | None) -> Flow[R]

Create an empty flow for None, otherwise emit the value once.

Parameters:

Name	Type	Description	Default
value	R | None	The optional item to emit; None produces an empty flow.	required

Returns:

Type	Description
Flow[R]	A flow containing value once, or no items when value is None.

iterate staticmethod

iterate(seed: R, function: Callable[[R], R]) -> Flow[R]

Emit seed, then repeatedly apply function to the previous value.

Parameters:

Name	Type	Description	Default
seed	R	The first value emitted or used to initialize the sequence.	required
function	Callable[[R], R]	Called with the previously emitted value to produce the next value.	required

Returns:

Type	Description
Flow[R]	A reusable, infinite flow beginning with seed.

generate staticmethod

generate(supplier: Callable[[], R]) -> Flow[R]

Create an infinite flow by calling supplier for each item.

Parameters:

Name	Type	Description	Default
supplier	Callable[[], R]	Called once for each requested item to produce that item.	required

Returns:

Type	Description
Flow[R]	A reusable, infinite flow of values returned by supplier.

unfold staticmethod

unfold(
    seed: C, function: Callable[[C], tuple[R, C] | None]
) -> Flow[R]

Generate values from state until the callback returns None.

Parameters:

Name	Type	Description	Default
seed	C	Initial state, reused by reference for each evaluation. Use an immutable seed or flow.defer when each evaluation needs fresh mutable state.	required
function	Callable[[C], tuple[R, C] | None]	Called with the current state and returns (value, next_state), or None to stop. A value inside the tuple may itself be None.	required

Returns:

Type	Description
Flow[R]	A lazy, reopenable flow. The seed is not emitted; taking a prefix does not call the function for subsequent values. Termination depends on the callback.

Raises:

Type	Description
TypeError	If function is not callable.

map

map(function: Callable[[T], R]) -> Flow[R]

Apply function to each item lazily.

map is lazy: the callable runs only when a terminal operation or iteration consumes the flow.

Parameters:

Name	Type	Description	Default
function	Callable[[T], R]	Receives each source item and returns its replacement value.	required

Returns:

Type	Description
Flow[R]	A lazy flow of mapped values. Any parallel settings on the current plan apply to this map.

map_parallel

map_parallel(
    function: Callable[[T], R],
    *,
    workers: int | None = None,
    backend: ParallelBackend = "thread",
    ordered: bool = True,
    buffer: int | None = None,
    executor: Executor | None = None,
) -> Flow[R]

Map items in a bounded thread or process pool.

Only a bounded number of tasks are submitted at once, preventing a slow consumer from creating unbounded work.

Parameters:

Name	Type	Description	Default
function	Callable[[T], R]	Receives one source item in a worker and returns its mapped value.	required
workers	int | None	Worker count for a new pool. Must be None with an external executor.	None
backend	ParallelBackend	thread or spawn-based process when creating a pool. An external executor determines its own backend.	'thread'
ordered	bool	Emit in source order when true, otherwise in completion order.	True
buffer	int | None	Maximum submitted futures retained before the pipeline waits for a result. Defaults to twice workers, or twice the CPU count when workers is None.	None
executor	Executor | None	Optional caller-owned executor to reuse. The query cancels queued work and waits for its own running callbacks before closing. It does not shut down this executor or wait for unrelated tasks.	None

Returns:

Type	Description
Flow[R]	A flow that submits bounded mapping work when consumed.

Raises:

Type	Description
ValueError	If workers or buffer is less than one, backend is unsupported, or workers and executor are both provided.
TypeError	If workers or buffer is not an integer, or executor is not a concurrent.futures.Executor.

parallel

parallel(
    *,
    workers: int | None = None,
    backend: ParallelBackend = "process",
    ordered: bool = True,
    buffer: int | None = None,
    executor: Executor | None = None,
) -> Flow[T]

Apply parallel settings to map operations added after this call.

Parameters:

Name	Type	Description	Default
workers	int | None	Worker count for a new pool. Must be None with an external executor.	None
backend	ParallelBackend	thread or spawn-based process when creating a pool. An external executor determines its own backend.	'process'
ordered	bool	Preserve source order for subsequent maps when true.	True
buffer	int | None	Maximum in-flight results retained by each subsequent map. Defaults to twice workers, or twice the CPU count when workers is None.	None
executor	Executor | None	Optional caller-owned executor borrowed by following maps. Each map cleans up only its own tasks and leaves the executor open.	None

Returns:

Type	Description
Flow[T]	A flow sharing this pipeline with parallel defaults for maps appended afterward.

sequential

sequential() -> Flow[T]

Return a flow whose following maps run sequentially.

Returns:

Type	Description
Flow[T]	A flow sharing this pipeline with parallel defaults cleared for later maps.

tap

tap(function: Callable[[T], None]) -> Flow[T]

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

Parameters:

Name	Type	Description	Default
function	Callable[[T], None]	Called for its side effect before each original item is emitted.	required

Returns:

Type	Description
Flow[T]	A flow that emits every original item unchanged after calling function.

filter

filter(predicate: Callable[[T], Any]) -> Flow[T]

Keep items for which predicate returns a truthy value.

filter is lazy and preserves encounter order; the predicate runs as items are requested.

Parameters:

Name	Type	Description	Default
predicate	Callable[[T], Any]	Called for each item; truthy results retain that item.	required

Returns:

Type	Description
Flow[T]	A flow containing only source items whose predicate result is truthy.

reject

reject(predicate: Callable[[T], Any]) -> Flow[T]

Drop items for which predicate returns a truthy value.

Parameters:

Name	Type	Description	Default
predicate	Callable[[T], Any]	Called for each item; truthy results drop that item.	required

Returns:

Type	Description
Flow[T]	A flow containing only source items whose predicate result is falsey.

compact

compact(selector: Selector | None = None) -> Flow[T]

Drop None values, optionally selected from each item.

Parameters:

Name	Type	Description	Default
selector	Selector | None	Optional callable, field name, index, path, or expression whose selected value is checked for None; without one, each item is checked directly.	None

Returns:

Type	Description
Flow[T]	A flow excluding items whose own or selected value is None. Other falsey values, such as 0 and False, are kept.

flat_map

flat_map(function: Callable[[T], Iterable[R]]) -> Flow[R]

Map each item to an iterable and emit the iterable contents.

Each input may emit zero or more output items. Nested iterables are consumed lazily in encounter order.

Parameters:

Name	Type	Description	Default
function	Callable[[T], Iterable[R]]	Maps each source item to the iterable whose contents are emitted.	required

Returns:

Type	Description
Flow[R]	A flow that lazily emits every mapped iterable in source order.

filter_map

filter_map(function: Callable[[T], R | None]) -> Flow[R]

Map items and discard results that are None.

Parameters:

Name	Type	Description	Default
function	Callable[[T], R | None]	Maps each source item to one output value or None.	required

Returns:

Type	Description
Flow[R]	A flow of non-None mapped results; falsey values such as 0 are retained.

pluck

pluck(selector: Selector) -> Flow[Any]

Select one field, index, attribute, or nested path from each item.

Parameters:

Name	Type	Description	Default
selector	Selector	Callable, field name, index, path, or expression evaluated for each output.	required

Returns:

Type	Description
Flow[Any]	A flow containing the value selected from each source item.

unique

unique() -> Flow[T]

Keep the first occurrence of each value in encounter order.

Returns:

Type	Description
Flow[T]	A flow containing the first occurrence of each distinct value.

distinct

distinct() -> Flow[T]

Keep the first occurrence of each value in encounter order.

Returns:

Type	Description
Flow[T]	The same lazy de-duplication pipeline produced by unique().

unique_by

unique_by(selector: Selector) -> Flow[T]

Keep the first item for each selected key.

Parameters:

Name	Type	Description	Default
selector	Selector	Callable, field name, index, path, or expression producing each uniqueness key.	required

Returns:

Type	Description
Flow[T]	A flow containing the first source item for each distinct selected key.

sort_by

sort_by(
    selector: Selector,
    *,
    reverse: bool = False,
    buffer_size: int | None = None,
    tempdir: str | PathLike[str] | None = None,
) -> Flow[T]

Sort items by a selector, optionally using bounded external storage.

Parameters:

Name	Type	Description	Default
selector	Selector	Callable, field name, index, path, or expression producing comparison keys.	required
reverse	bool	Emit items in descending selected-key order when true.	False
buffer_size	int | None	Items per sorted in-memory run; None performs one in-memory sort.	None
tempdir	str | PathLike[str] | None	Directory for temporary run files when buffer_size is set.	None

Returns:

Type	Description
Flow[T]	A flow that emits every source item ordered by its selected value.

sorted

sorted(
    *,
    key: Callable[[T], Any] | None = None,
    reverse: bool = False,
    buffer_size: int | None = None,
    tempdir: str | PathLike[str] | None = None,
) -> Flow[T]

Sort the flow, optionally using bounded external runs.

Parameters:

Name	Type	Description	Default
key	Callable[[T], Any] | None	Optional callable used to derive each item's comparison key.	None
reverse	bool	Emit items in descending comparison order when true.	False
buffer_size	int | None	Items per sorted in-memory run; None performs one in-memory sort.	None
tempdir	str | PathLike[str] | None	Directory for temporary run files when buffer_size is set.	None

Returns:

Type	Description
Flow[T]	A flow that globally orders the source by key or by the items themselves.

external_sort

external_sort(
    *,
    key: Callable[[T], Any] | None = None,
    reverse: bool = False,
    buffer_size: int = 100000,
    tempdir: str | PathLike[str] | None = None,
) -> Flow[T]

Sort with a bounded in-memory buffer and temporary files.

Sorted runs are written to temporary files and merged lazily, keeping peak in-memory items bounded.

Parameters:

Name	Type	Description	Default
key	Callable[[T], Any] | None	Optional callable used to derive each item's comparison key.	None
reverse	bool	Emit items in descending comparison order when true.	False
buffer_size	int	Maximum items sorted in memory for each temporary run.	100000
tempdir	str | PathLike[str] | None	Directory in which temporary sorted runs are created.	None

Returns:

Type	Description
Flow[T]	A globally sorted flow produced by lazily merging bounded temporary runs.

external_sort_by

external_sort_by(
    selector: Selector,
    *,
    reverse: bool = False,
    buffer_size: int = 100000,
    tempdir: str | PathLike[str] | None = None,
) -> Flow[T]

Sort by a selector using bounded memory and temporary files.

Sorted runs are written to temporary files and merged lazily, keeping peak in-memory items bounded.

Parameters:

Name	Type	Description	Default
selector	Selector	Callable, field name, index, path, or expression producing comparison keys.	required
reverse	bool	Emit items in descending selected-key order when true.	False
buffer_size	int	Maximum items sorted in memory for each temporary run.	100000
tempdir	str | PathLike[str] | None	Directory in which temporary sorted runs are created.	None

Returns:

Type	Description
Flow[T]	A flow ordered by the selected value using bounded temporary runs.

chunk

chunk(
    size: int, *, strict: bool = False
) -> Flow[tuple[T, ...]]

Group consecutive items into fixed-size tuples.

Parameters:

Name	Type	Description	Default
size	int	Maximum number of consecutive items in each tuple.	required
strict	bool	Raise during consumption if the final chunk is shorter than size.	False

Returns:

Type	Description
Flow[tuple[T, ...]]	A flow of non-overlapping tuples. By default, a final shorter tuple is emitted.

Raises:

Type	Description
TypeError	If size does not support the integer index protocol.
ValueError	If size is less than one, or a final chunk is incomplete in strict mode.

map_batches

map_batches(
    function: Callable[[tuple[T, ...]], R], *, size: int
) -> Flow[R]

Apply function to consecutive tuples and emit one result per batch.

Batches are created lazily and include a shorter final batch. Empty input does not call function. Each returned value stays intact; use flat_map(iter) explicitly when the callback returns iterables whose contents should be emitted.

Parallel defaults configured with parallel() apply to the batch callback. Its buffer counts batches, so at most size * buffer input items are submitted at once. Process callbacks and their inputs and results must be picklable.

Parameters:

Name	Type	Description	Default
function	Callable[[tuple[T, ...]], R]	Receives a tuple of up to size consecutive items and returns one result.	required
size	int	Maximum number of input items passed to each callback.	required

Returns:

Type	Description
Flow[R]	A lazy flow of batch results, using the current map parallel settings.

Raises:

Type	Description
TypeError	If size does not support the integer index protocol.
ValueError	If size is less than one.

batch_by_size

batch_by_size(
    max_size: int,
    *,
    max_count: int | None = None,
    get_size: Callable[[T], int] = _default_item_size,
    strict: bool = True,
) -> Flow[tuple[T, ...]]

Build batches constrained by item count and total measured size.

Parameters:

Name	Type	Description	Default
max_size	int	Maximum sum of item sizes in a normal batch.	required
max_count	int | None	Optional maximum item count per batch.	None
get_size	Callable[[T], int]	Returns a non-negative integer size for each item; defaults to len.	_default_item_size
strict	bool	Raise when one item exceeds max_size; when false, emit it in an oversized singleton batch.	True

Returns:

Type	Description
Flow[tuple[T, ...]]	A flow of non-empty tuple batches within the size and count limits. With strict=False, an item larger than max_size is emitted in its own batch.

Raises:

Type	Description
TypeError	If a bound does not support the integer index protocol.
ValueError	If a bound is less than one.

window

window(size: int, *, step: int = 1) -> Flow[tuple[T, ...]]

Emit sliding tuples of size with the requested step.

Parameters:

Name	Type	Description	Default
size	int	Number of items in each full window.	required
step	int	Number of source items consumed between successive windows.	1

Returns:

Type	Description
Flow[tuple[T, ...]]	A flow of full sliding windows. A non-empty source shorter than size produces one partial window; otherwise, partial trailing windows are omitted.

Raises:

Type	Description
TypeError	If size or step does not support the integer index protocol.
ValueError	If size or step is less than one.

window_reduce

window_reduce(
    size: SupportsIndex,
    combine: Callable[[T, T], T],
    *,
    step: SupportsIndex = 1,
) -> Flow[T]

Reduce sliding windows while reusing ordered partial aggregates.

Parameters:

Name	Type	Description	Default
size	SupportsIndex	Number of source items in each full window.	required
combine	Callable[[T, T], T]	Associative synchronous function that never mutates its arguments. Source values and returned aggregates must remain unchanged during iteration.	required
step	SupportsIndex	Number of source items between successive window starts.	1

Returns:

Type	Description
Flow[T]	A flow of reduced windows, in encounter order. Empty input emits nothing; input shorter than size emits one partial result. After a full window, incomplete trailing windows are omitted, following window.

Raises:

Type	Description
TypeError	If size or step does not support the integer index protocol.
ValueError	If size or step is less than one.

Notes

Grouping of combine calls differs from a left fold. Floating-point addition is not associative. Use window().map(...) for arbitrary reducers or side effects. Combine may run while filling a window that is never emitted, so its errors can surface before the corresponding window boundary. The queue caches at most size + 1 entries, whose byte sizes may grow. Combine calls are amortized constant per item; a stack transfer can take linear work in size.

group_runs

group_runs(
    key: Selector | None = None,
) -> Flow[tuple[T, ...]]

Group consecutive items that share the same key.

Parameters:

Name	Type	Description	Default
key	Selector | None	Optional selector for run identity; adjacent items themselves are compared when omitted.	None

Returns:

Type	Description
Flow[tuple[T, ...]]	A flow of non-empty tuples, one for each contiguous run of equal keys.

pairwise

pairwise() -> Flow[tuple[T, T]]

Emit each adjacent pair of items.

Returns:

Type	Description
Flow[tuple[T, T]]	A flow of overlapping (previous, current) pairs; fewer than two items emit nothing.

pair_map

pair_map(function: Callable[[T, T], R]) -> Flow[R]

Apply a two-argument function to each adjacent pair.

Parameters:

Name	Type	Description	Default
function	Callable[[T, T], R]	Called as function(previous, current) for each adjacent pair.	required

Returns:

Type	Description
Flow[R]	A flow containing one mapped result per adjacent source pair.

enumerate

enumerate(start: int = 0) -> Flow[tuple[int, T]]

Pair each item with a consecutive index starting at start.

Parameters:

Name	Type	Description	Default
start	int	Integer index paired with the first source item.	0

Returns:

Type	Description
Flow[tuple[int, T]]	A flow of (index, item) pairs with consecutive integer indices.

zip

zip(
    other: Iterable[U], *, strict: bool = False
) -> Flow[tuple[T, U]]

Pair items with another iterable until one side ends.

Parameters:

Name	Type	Description	Default
other	Iterable[U]	Synchronous iterable providing the right-hand item in each pair.	required
strict	bool	Raise ValueError during consumption when the two inputs have different lengths.	False

Returns:

Type	Description
Flow[tuple[T, U]]	A flow of pairs ending with the shorter input unless strict is true.

zip_longest

zip_longest(
    other: Iterable[U], *, fillvalue: Any = None
) -> Flow[tuple[T | Any, U | Any]]

Pair with another iterable until both sides end, filling missing values.

Parameters:

Name	Type	Description	Default
other	Iterable[U]	Synchronous iterable providing right-hand values.	required
fillvalue	Any	Substitute used on whichever input is exhausted first.	None

Returns:

Type	Description
Flow[tuple[T | Any, U | Any]]	A flow of pairs whose length matches the longer input.

intersperse

intersperse(separator: T) -> Flow[T]

Insert separator between consecutive items.

Parameters:

Name	Type	Description	Default
separator	T	Item emitted once between each pair of adjacent source items.	required

Returns:

Type	Description
Flow[T]	A flow with separator between source items and never at either boundary.

merge_sorted

merge_sorted(
    other: Iterable[T] | Flow[T],
    *,
    key: Selector | None = None,
) -> Flow[T]

Stably merge two ascending inputs, returning the original objects.

Keys must have one fixed exact builtin int, str, bytes, or flat tuple shape. Checks cover consumed rows. Equal keys from the left are emitted first. Each input has at most one lookahead row; inputs are never sorted here.

Parameters:

Name	Type	Description	Default
other	Iterable[T] | Flow[T]	Another sorted iterable or Flow.	required
key	Selector | None	Shared selector, or None to compare the items themselves.	None

Returns:

Type	Description
Flow[T]	A lazy two-input Flow. Closing it closes both owned input iterators.

concat

concat(*others: Iterable[T]) -> Flow[T]

Emit this flow followed by each supplied iterable.

Reused cursors are closed once per execution. Cursors that do not support weak references can remain reachable until that execution finishes.

Parameters:

Name	Type	Description	Default
*others	Iterable[T]	Synchronous iterables opened and emitted after this flow, in argument order.	()

Returns:

Type	Description
Flow[T]	A flow that drains this source and then each additional iterable in order.

cross

cross(
    other: Iterable[U], *, max_right: int | None = None
) -> Flow[tuple[T, U]]

Buffer another iterable once and emit a left-major Cartesian product.

Parameters:

Name	Type	Description	Default
other	Iterable[U]	Synchronous iterable buffered as the right side after the first left item.	required
max_right	int | None	Optional maximum number of right-side items that may be buffered.	None

Returns:

Type	Description
Flow[tuple[T, U]]	A flow of (left, right) pairs with every right item repeated for each left item.

Raises:

Type	Description
BufferLimitError	During consumption if other contains more than max_right items.

rolling_statistics

rolling_statistics(
    size: SupportsIndex,
    *,
    selector: Selector | None = None,
    ddof: SupportsIndex = 0,
) -> Flow[tuple[int, float | None, float | None]]

Emit count, mean, and variance for the most recent input items.

Parameters:

Name	Type	Description	Default
size	SupportsIndex	Positive integer window width, counted in source items.	required
selector	Selector | None	Synchronous numeric selector; None selects each whole item. A selected None occupies a window slot but does not contribute to the statistics.	None
ddof	SupportsIndex	Nonnegative integer subtracted from the valid count for variance.	0

Returns:

Type

Description

Flow[tuple[int, float | None, float | None]]

A flow of (count, mean, variance) tuples, including warmup windows. Undefined statistics are None. Each evaluation owns fresh state and retains at most size converted values. Selected numbers must be finite; a compensated mean and shifted Welford variance updates are periodically rebuilt to limit drift. Floating-point overflow and rounding remain possible. Rebuilds take O(size) time; regular updates take O(1), with additional rebuilds after severe cancellation.

scan

scan(initial: R, function: Callable[[R, T], R]) -> Flow[R]

Emit each left-to-right accumulator state after consuming one item.

Unlike reduce, scan emits every intermediate accumulator state.

Parameters:

Name	Type	Description	Default
initial	R	Accumulator passed to the first callback; it is not emitted by itself.	required
function	Callable[[R, T], R]	Called as function(state, item) to produce each next state.	required

Returns:

Type	Description
Flow[R]	A flow with one accumulated state for every source item.

map_accum

map_accum(
    initial: C, function: Callable[[C, T], tuple[C, R]]
) -> Flow[R]

Map items in encounter order while threading a separate state.

Parameters:

Name	Type	Description	Default
initial	C	State passed to the first callback, reused by reference for each evaluation. Use flow.defer if mutable state must be freshly allocated.	required
function	Callable[[C, T], tuple[C, R]]	Called as function(state, item) and returns (next_state, output). Callbacks run sequentially even when the flow has parallel map defaults.	required

Returns:

Type	Description
Flow[R]	A lazy flow with one output per input, including None outputs. Empty input emits nothing; the final state is not emitted. State holders are independent for concurrent or repeated evaluations, as with scan.

Raises:

Type	Description
TypeError	If function is not callable.

scan_right

scan_right(
    initial: R,
    function: Callable[[T, R], R],
    *,
    max_items: int | None = None,
) -> Flow[R]

Buffer the source, accumulate from right to left, and emit states in source order.

Parameters:

Name	Type	Description	Default
initial	R	Accumulator passed to the rightmost callback; it is not emitted by itself.	required
function	Callable[[T, R], R]	Called as function(item, state) from the rightmost item to the leftmost.	required
max_items	int | None	Optional maximum number of source items that may be buffered.	None

Returns:

Type	Description
Flow[R]	A flow with one right-fold state per source item, ordered like the original items.

Raises:

Type	Description
BufferLimitError	During consumption if the source exceeds max_items.

gather

gather(gatherer: Gatherer[T, Any, R]) -> Flow[R]

Apply a stateful Gatherer that may emit zero or more values per item.

A gatherer may retain state and emit zero, one, or many outputs for each input item.

Parameters:

Name	Type	Description	Default
gatherer	Gatherer[T, Any, R]	The stateful gatherer applied to this pipeline.	required

Returns:

Type	Description
Flow[R]	A flow of values emitted by gatherer while it integrates and finishes the source.

fold

fold(
    initializer: Callable[[], R],
    function: Callable[[R, T], R],
) -> Flow[R]

Consume all items into one emitted value using fresh state per iteration.

Parameters:

Name	Type	Description	Default
initializer	Callable[[], R]	Called once per evaluation to create the initial accumulator.	required
function	Callable[[R, T], R]	Called as function(state, item) and returns the replacement accumulator.	required

Returns:

Type	Description
Flow[R]	A flow that emits exactly one final accumulator, including for an empty source.

Notes

Annotate the resulting Flow[R] to declare the accumulator type and give a lambda step its parameter context. Every source item invokes the step, including the first item. Neither state nor callback results are awaited. Any can bypass the static contract; state values are not validated at runtime.

prepend

prepend(*values: T) -> Flow[T]

Emit values before the items from this flow.

Parameters:

Name	Type	Description	Default
*values	T	Items to emit before the first source item, in argument order.	()

Returns:

Type	Description
Flow[T]	A flow containing values followed by every source item.

append

append(*values: T) -> Flow[T]

Emit values after the items from this flow.

Parameters:

Name	Type	Description	Default
*values	T	Items to emit after the source completes, in argument order.	()

Returns:

Type	Description
Flow[T]	A flow containing every source item followed by values.

map_first

map_first(function: Callable[[T], T]) -> Flow[T]

Transform only the first item, if one exists.

Parameters:

Name	Type	Description	Default
function	Callable[[T], T]	Maps the first item; it is never called for an empty source.	required

Returns:

Type	Description
Flow[T]	A flow with only its first item replaced by function(first).

map_last

map_last(function: Callable[[T], T]) -> Flow[T]

Transform only the last item, if one exists.

Parameters:

Name	Type	Description	Default
function	Callable[[T], T]	Maps the final item; it is never called for an empty source.	required

Returns:

Type	Description
Flow[T]	A flow with only its final item replaced by function(last).

collapse

collapse(
    collapsible: Callable[[T, T], bool],
    merger: Collector[T, C, R],
) -> Flow[R]

collapse(
    collapsible: Callable[[T, T], bool],
    merger: Callable[[T, T], T],
) -> Flow[T]

collapse(
    collapsible: Callable[[T, T], bool],
    merger: Callable[[T, T], T] | Collector[T, C, R],
) -> Flow[T] | Flow[R]

Merge adjacent items while collapsible returns true.

Parameters:

Name	Type	Description	Default
collapsible	Callable[[T, T], bool]	Called on neighboring original items to decide whether a run continues.	required
merger	Callable[[T, T], T] | Collector[T, C, R]	Binary function combining the current aggregate with the next item, or a Collector with synchronous lifecycle hooks.	required

Returns:

Type	Description
Flow[T] | Flow[R]	A flow containing one merged or finished result per contiguous run. Empty input emits nothing and does not initialize a collector.

Notes

A collector gets fresh state per nonempty run. Its done hook stops further steps for that run, but all adjacent predicates still run to find the boundary. One boundary item is read ahead; the next run is initialized only when consumption resumes. The final run is finished only after normal exhaustion. Collector hooks run while reading input, and their errors can precede errors that a buffered group would encounter first. No raw run buffer is created; state size depends on the collector. Combine and native metadata are unused. A finisher returning an awaitable emits that object without awaiting it.

attempt

attempt(function: Callable[[T], R]) -> Flow[Result[R]]

Map each item and wrap success or failure in a Result.

Parameters:

Name	Type	Description	Default
function	Callable[[T], R]	Maps one source item and may raise an Exception.	required

Returns:

Type	Description
Flow[Result[R]]	A flow of Ok mapped values and Err objects for raised exceptions.

with_engine

with_engine(engine: Engine) -> Flow[T]

Request automatic, Python, or native execution for this plan.

Selecting an engine changes execution policy without consuming the source.

Parameters:

Name	Type	Description	Default
engine	Engine	The execution engine requested for this pipeline.	required

Returns:

Type	Description
Flow[T]	An equivalent lazy flow whose plan requests engine during execution.

explain

explain(
    terminal: TerminalName = "iterate",
) -> PlanExplanation

Describe engine selection, stages, and fused operations without executing.

This inspection method does not consume or execute the source.

Parameters:

Name	Type	Description	Default
terminal	TerminalName	Terminal operation to include when validating and selecting the engine.	'iterate'

Returns:

Type	Description
PlanExplanation	A structured explanation of the selected engine and planned stages.

pairs

pairs() -> fpstreams.Pairs[K, V]

pairs() -> fpstreams.Pairs[Any, Any]

pairs() -> fpstreams.Pairs[Any, Any]

View a flow of two-tuples as a key/value Pairs pipeline.

Returns:

Type	Description
Pairs[Any, Any]	A lazy Pairs view over this flow.

rows

rows() -> fpstreams.Rows[T]

View this flow as a lazy record pipeline without inspecting its items.

Returns:

Type	Description
Rows[T]	A Rows view that shares this flow's plan and source ownership.

select

select(
    *selectors: str | int, **named: Selector
) -> fpstreams.Rows[dict[str, Any]]

Project record fields through a lazy Rows view of this flow.

with_columns

with_columns(
    **columns: Selector,
) -> fpstreams.Rows[dict[str, Any]]

Add or replace record fields through a lazy Rows view of this flow.

rename

rename(**columns: str) -> fpstreams.Rows[dict[str, Any]]

Rename record fields through a lazy Rows view of this flow.

cast

cast(
    **columns: Callable[[Any], Any],
) -> fpstreams.Rows[dict[str, Any]]

Convert named record fields through a lazy Rows view of this flow.

fill_nulls

fill_nulls(
    **replacements: object,
) -> fpstreams.Rows[dict[str, Any]]

Replace missing or None record fields through a lazy Rows view.

drop_nulls

drop_nulls(
    *selectors: Selector, how: Literal["any", "all"] = "any"
) -> fpstreams.Rows[T]

Drop records according to selected None values through a Rows view.

explode

explode(
    selector: Selector,
    *,
    into: str | None = None,
    outer: bool = False,
) -> fpstreams.Rows[dict[str, Any]]

Expand one selected iterable through a lazy Rows view.

unnest

unnest(
    column: str, *, prefix: str = ""
) -> fpstreams.Rows[dict[str, Any]]

Promote fields from one nested record through a lazy Rows view.

unpivot

unpivot(
    *columns: str,
    names_to: str = "variable",
    values_to: str = "value",
) -> fpstreams.Rows[dict[str, Any]]

Reshape wide records into name/value records through a Rows view.

pivot

pivot(
    *,
    index: Selector | tuple[Selector, ...],
    columns: Selector,
    values: Selector,
    aggregate: str | Callable[[Any, Any], Any] = "error",
    fill: Any = None,
) -> fpstreams.Rows[dict[str, Any]]

Reshape long records into wide records through a lazy Rows view.

group_by

group_by(
    *selectors: Selector, **named: Selector
) -> fpstreams.tabular.GroupedRows[T]

Describe grouped aggregation through a lazy Rows view of this flow.

take

take(count: SupportsIndex) -> Flow[T]

Emit at most count items, then close the upstream iterator.

Parameters:

Name	Type	Description	Default
count	SupportsIndex	Maximum number of leading items to emit.	required

Returns:

Type	Description
Flow[T]	A flow containing only the first count source items.

Raises:

Type	Description
TypeError	If count does not support the integer index protocol.
ValueError	If count is negative.

limit

limit(count: SupportsIndex) -> Flow[T]

Emit at most count items; alias of take.

Parameters:

Name	Type	Description	Default
count	SupportsIndex	Maximum number of leading items to emit.	required

Returns:

Type	Description
Flow[T]	The same bounded pipeline produced by take(count).

drop

drop(count: SupportsIndex) -> Flow[T]

Skip count items before yielding the remainder.

Parameters:

Name	Type	Description	Default
count	SupportsIndex	Number of leading items to consume without emitting.	required

Returns:

Type	Description
Flow[T]	A flow containing every source item after the first count.

Raises:

Type	Description
TypeError	If count does not support the integer index protocol.
ValueError	If count is negative.

skip

skip(count: SupportsIndex) -> Flow[T]

Skip count items; alias of drop.

Parameters:

Name	Type	Description	Default
count	SupportsIndex	Number of leading items to consume without emitting.	required

Returns:

Type	Description
Flow[T]	The same suffix pipeline produced by drop(count).

take_while

take_while(predicate: Callable[[T], bool]) -> Flow[T]

Emit the longest prefix that satisfies predicate.

Parameters:

Name	Type	Description	Default
predicate	Callable[[T], bool]	Called on leading items until its first falsey result.	required

Returns:

Type	Description
Flow[T]	A flow ending before the first source item whose predicate result is falsey.

take_while_inclusive

take_while_inclusive(
    predicate: Callable[[T], bool],
) -> Flow[T]

Emit through the first item that fails predicate.

Parameters:

Name	Type	Description	Default
predicate	Callable[[T], bool]	Called on leading items through its first falsey result.	required

Returns:

Type	Description
Flow[T]	A flow ending after emitting the first item whose predicate result is falsey.

drop_while

drop_while(predicate: Callable[[T], bool]) -> Flow[T]

Skip the longest prefix that satisfies predicate.

Parameters:

Name	Type	Description	Default
predicate	Callable[[T], bool]	Called only while leading items produce truthy results.	required

Returns:

Type	Description
Flow[T]	A flow starting with the first item for which predicate is falsey. Remaining items pass through without further predicate calls.

Merging sorted inputs

merge_sorted(other, key=None) checks ascending keys while consuming two inputs. Keys must be exact int, str, bytes, or nonempty flat tuples with one fixed type per position across both inputs. Equal keys from the left come first; output objects are the original inputs. Each side has one lookahead row.

from fpstreams import flow

assert flow([1, 3]).merge_sorted([1, 2]).to_list() == [1, 1, 2, 3]

Added in 2.2.0. It uses Python in auto mode; forcing native is unsupported. Iterator aliasing is rejected before the first pull, and both owned inputs are closed on errors or early termination. Nested merges keep one pair of cursors per merge; they are not a heap-based k-way merge.

Sorted paired operations acquire their branch source iterators left to right, without pulling rows, before either input runs. This includes source leaves in composed branches, so a shared underlying iterator can be rejected before it is read. Each acquired iterator closes once, even if a branch never yields a row. An outer take(0) does not start this acquisition phase.

File exports

to_jsonl() writes one JSON value per line, including scalars and records. Use default for custom serialization and ensure_ascii to control Unicode escaping. The source is consumed once and the output is written incrementally.

to_csv(), to_json(), and to_jsonl() accept atomic=True to preserve the old target until writing and source cleanup succeed. if_exists="error" requires atomic output and prevents overwriting a concurrently created target. See atomic file output for examples and filesystem limits. These additions were introduced in 2.2.0.