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.

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 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

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 with query column pruning
`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,
) -> 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 so spreadsheet software treats them as text.	`False`

to_json

to_json(
    path: str | PathLike[str],
    *,
    encoding: str = "utf-8",
    ensure_ascii: bool = False,
    default: Callable[[Any], Any] | None = None,
) -> 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`

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

Compute several named aggregations while traversing the flow once.

All named aggregators are updated during the same source traversal.

Parameters:

Name	Type	Description	Default
`**aggregations`	`Aggregator`	Result names mapped to aggregators updated in one traversal.	`{}`

Returns:

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

collect

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() -> Any

View this synchronous pipeline as an AsyncFlow.

Returns:

Type	Description
`Any`	An `AsyncFlow` that yields the same items.

first

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(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], 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, 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`.

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,
) -> 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, or `None` to use the executor's default.	`None`
`backend`	`ParallelBackend`	Run callbacks in a `thread` or spawn-based `process` pool.	`'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.	`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, or backend is unsupported.

parallel

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

Apply parallel settings to map operations added after this call.

Parameters:

Name	Type	Description	Default
`workers`	`int \| None`	Worker count, or `None` to use the executor's default.	`None`
`backend`	`ParallelBackend`	Run callbacks in a `thread` or spawn-based `process` pool.	`'process'`
`ordered`	`bool`	Preserve source order for subsequent maps when true.	`True`
`buffer`	`int \| None`	Maximum in-flight results retained by each subsequent map.	`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 equal to 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) -> 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

Returns:

Type	Description
`Flow[tuple[T, ...]]`	A flow of non-overlapping tuples, including a final shorter tuple when needed.

Raises:

Type	Description
`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.

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
`ValueError`	If size or step is less than one.

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.

concat

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

Emit this flow followed by each supplied iterable.

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.

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.

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.

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: Callable[[T, T], T],
) -> Flow[T]

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]`	Combines the current run aggregate with the next item.	required

Returns:

Type	Description
`Flow[T]`	A flow containing one merged aggregate for each contiguous collapsible run.

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() -> Any

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

Returns:

Type	Description
`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: int) -> Flow[T]

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

Parameters:

Name	Type	Description	Default
`count`	`int`	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
`ValueError`	If count is negative.

limit

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

Emit at most count items; alias of take.

Parameters:

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

Returns:

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

drop

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

Skip count items before yielding the remainder.

Parameters:

Name	Type	Description	Default
`count`	`int`	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
`ValueError`	If count is negative.

skip

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

Skip count items; alias of drop.

Parameters:

Name	Type	Description	Default
`count`	`int`	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.

Flow

Record operations and the Rows view

Tabular source routing

Explaining terminal execution

CSV safety

Creating a flow

Methods

fpstreams.Flow

run_with_report

to_list

to_tuple

to_set

to_pandas

to_numpy

to_csv

to_json

describe

aggregate

collect

join

for_each

partition

partition_results

to_async

first

last

find

find_index

index_of

nth

count

sum

mean

variance

std

min

max

top

bottom

minmax

any

all

none

reduce

reduce_right

reduce_by

frequencies

of staticmethod

from_iterable staticmethod

empty staticmethod

of_nullable staticmethod

iterate staticmethod

generate staticmethod

map

map_parallel

parallel

sequential

tap

filter

reject

compact

flat_map

filter_map

pluck

unique

distinct

unique_by

sort_by

sorted

external_sort

external_sort_by

chunk

batch_by_size

window

group_runs

pairwise

pair_map

enumerate

zip

zip_longest

of `staticmethod`

from_iterable `staticmethod`

empty `staticmethod`

of_nullable `staticmethod`

iterate `staticmethod`

generate `staticmethod`