Skip to content

AsyncFlow

AsyncFlow[T] combines async iteration with bounded concurrent mapping, merging, rate control, latest-request switching, timeouts, and buffer-by-time operations.

Use aflow(source) with an iterable or async iterable.

Consumption and cleanup

In the 2.5.0 async uniqueness path, distinct and distinct_by compare hashable and unhashable keys across categories. The first accepted equal key wins; additional equality callbacks and their errors are observable. State uses O(H + U) references and cross-category comparisons can take O(H × U) time for H hashable and U unhashable keys. A new hashable key is compared as seen == key against prior unhashable keys; a new unhashable key is compared as key == seen against prior hashable keys in encounter order.

Async group_runs reads the first item of the next group and resolves its key before emitting the previous group as a complete tuple. A key failure at that boundary occurs before the previous group is emitted. It holds the whole current group in memory and keeps original item references in the tuple. An infinite run with no key change cannot produce a completed group.

In the 2.5.0 async concat path, an additional source is opened only when reached and is closed on completion or early exit. A directly supplied one-shot iterator shared with the input keeps one owner, including when it is a synchronous iterator adapted for async iteration.

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

In the 2.5.0 source, async chunk(size, strict=True) reads an incomplete tail before raising ValueError and does not emit that tail. The default emits the shorter tuple. Stopping before the tail does not validate it; when the error occurs, the operation closes the owned source when it supports closing. Async batch_by_size checks the next item and its size before emitting a pending batch, including when max_count has been reached. A source error, size error, or strict oversized item can prevent that pending batch from being emitted. Normal exhaustion emits a nonempty tail.

Async zip and zip_longest await their left and right inputs in order, not concurrently. A left input that waits for the right input to start can stall. The right input can end after the left has already produced an extra item. Strict zip validates unequal lengths only as far as it is consumed. zip_longest defaults to None padding; use a unique sentinel to distinguish padding from a real None. A mutable fill value is reused by reference. combine_latest cannot emit until every source has a value; when a source ends empty, the other pending sources are cancelled and closed.

Direct async iteration keeps the output iterator open after break if the caller retains it. Close it explicitly when stopping early:

from contextlib import aclosing
from fpstreams import aflow

async def first_three():
    async with aclosing(aiter(aflow(range(10)))) as values:
        async for value in values:
            if value == 2:
                break

When a downstream take(1) yields its last item through direct iteration, cleanup may wait until the iterator is resumed or explicitly closed. Terminal methods manage their iterator when they finish. An aflow backed directly by an async generator closes that owned generator when first() stops. A later concat source that has not been opened is not closed merely because an earlier source short-circuits. merge_map inner sources can outlive the outer source; manage any external resources they depend on for the needed lifetime.

Async scan has the same state reference and replay rules as synchronous scan; an empty source emits no state. scan_right buffers and folds the full source before emitting. Its max_items check reads one more item when detecting an overflow, just as in the synchronous path. Use aflow.defer to create the pipeline and mutable seed anew for each execution.

For a direct aflow.defer(factory) source, take(0).to_list() opens and closes the iterator without pulling it, whereas collect(Collectors.head(0)) does not call the factory. A factory error still propagates in the first case. intersperse reads the next item before emitting its separator. delay opens the source before its initial wait but does not pull during that wait; throttle may already have pulled an item while waiting to emit it. A prefix result does not validate unread items or errors.

Flattening mapped iterables

flat_map() expands each mapped iterable in order, accepting either a synchronous or asynchronous iterable. A string expands into characters and a dictionary into keys; wrap the value in another iterable to keep it intact.

The serial implementation manages each acquired inner iterator and attempts its close() or aclose() when supported, including on exhaustion, failure, or downstream early termination. If a mapper returns an iterator object that returns itself from __iter__ or __aiter__, the current implementation calls that method once to acquire it and again when entering the iteration loop. Side effects in these methods can therefore be observed twice.

Statistics and rolling windows (2.5.0)

rolling_statistics(size, selector=..., ddof=0) returns an AsyncFlow of (count, mean, variance) tuples. The selector remains synchronous. Missing values, warmup output, configuration errors, and fresh-state rules match Flow statistics.

import asyncio
from fpstreams import Collectors, aflow


async def summarize():
    summary = await aflow([2, None, 4, 6]).collect(Collectors.statistics())
    windows = await aflow([2, None, 4, 6]).rolling_statistics(2).to_list()
    assert summary["mean"] == 4.0
    assert windows[-1] == (2, 5.0, 1.0)


asyncio.run(summarize())

Collector state callbacks follow their synchronous contract; await the async terminal that consumes the flow. Cancellation closes the owned upstream iterator. The optional sketches use the same collect() entry point; see Collector statistics. These additions require the source version.

Creating an async flow

Call Behavior
aflow(source) Wrap a synchronous or asynchronous iterable
aflow.defer(factory) Call the source factory for each execution
aflow.unfold(seed, function) (2.4.0) Resolve (value, next_state) transitions until None is returned
aflow.from_queue(queue, stop=...) Read a caller-owned asyncio.Queue once, optionally until an identity sentinel
aflow.from_file(path) Read text lines without blocking the event loop
aflow.interval(seconds) Emit increasing integers on a timer
aflow.paginate(fetch_page) Fetch pages until the returned cursor is None

Queue sources do not call task_done() and do not own the producer or queue. Avoid prefetch() when Queue.join() or per-item acknowledgements must track each get() exactly.

File sources own their handles and use the optional async extra (aiofiles). In the 2.5.0 source, cancellation waits for in-flight file operations and any required close attempt before propagating. If close fails, cancellation is preserved and the close failure is added to its notes. Cancelling an await does not stop a filesystem call running in a thread: a stalled call can delay cancellation indefinitely. Closing the file does not shut down the event loop's shared executor.

Reduce values and check their types

await flow.reduce(function, initial) reduces from left to right. The callback may return a value or an awaitable; an awaitable callback result is awaited once. An empty flow returns the supplied initial object unchanged. Without an initial value, a singleton returns its item unchanged, even if that item is awaitable; an empty flow raises EmptyFlowError.

reduce_right passes (item, accumulator) to the callback and optionally limits buffering with max_items; fold_right is its alias. These terminal methods currently return Any to type checkers after awaiting, so an annotated callback alone does not narrow the result. See the synchronous reduce type examples for a caller-owned runtime check. fold, by contrast, returns a lazy AsyncFlow and resolves its initializer factory result once when the flow is consumed.

import asyncio
from fpstreams import aflow


async def main():
    seed = asyncio.get_running_loop().create_future()
    seed.set_result(10)
    assert await aflow([]).reduce(lambda total, item: total + item, seed) is seed
    assert await aflow([1]).reduce(lambda total, item: seed, 0) == 10


asyncio.run(main())

Explicit asynchronous accumulator contract (2.5.0)

fold_async(initializer, function) returns a lazy AsyncFlow[R]. Both hooks must return awaitables under this typed contract. Each hook result is awaited exactly once, and its resolved value becomes the state. Annotate the resulting flow to declare the state contract and provide context for ordinary lambda callbacks:

import asyncio
from fpstreams import AsyncFlow, aflow


async def initial_text() -> str:
    return ""


async def append_text(state: str, item: int) -> str:
    return state + str(item)


async def main():
    query: AsyncFlow[str] = aflow([1, 2]).fold_async(
        initial_text, lambda state, item: append_text(state, item)
    )
    assert await query.first() == "12"


asyncio.run(main())

This explicit path keeps the existing reduce and mixed-hook fold interfaces. It does not improve the old reduce return declaration. An empty source emits the resolved initializer state. Every source item invokes the step, including a singleton, so migration from unseeded reduce requires an explicit initial state. Each evaluation calls the initializer again.

The state R may itself be awaitable. For example, an async def initializer annotated -> asyncio.Future[int] can return a Future as state, and an async def step with the same return annotation can retain it. The hook coroutines are awaited once; the Future is neither awaited again nor replaced, even on empty or singleton inputs. A synchronous initializer returning the Future directly belongs to the existing mixed-hook fold, which resolves that Future and uses its value as state. Wrap the initializer and step in async functions when the Future object itself is the intended state.

As with other Python type annotations, Any can bypass this explicit contract. Execution reuses fold; it does not add runtime checks for hook awaitability or state types, and nested Any remains unknown. Use a caller-owned runtime check when accepting untrusted dynamic callback results.

Thread state through async callbacks (2.4.0)

AsyncFlow.unfold and aflow.unfold accept synchronous, asynchronous, or mixed callbacks. Each call resolves before the next transition starts. Returning (None, next_state) emits None; returning None ends the source.

map_accum(initial, function) consumes an existing source in encounter order. The callback resolves to (next_state, output), and only the output is emitted: an awaitable transition is awaited once, then an awaitable output is also awaited once. The next state is passed through without awaiting it. Synchronous Flow.map_accum() instead emits its output unchanged, including an awaitable. When a task is cancelled while awaiting a native Future output, that Future is normally cancelled too; asyncio.shield() can protect an underlying Future from this cancellation. Other cancellation sources can still cancel it.

import asyncio
from fpstreams import aflow


async def countdown(state: int) -> tuple[int, int] | None:
    await asyncio.sleep(0)
    return (state, state - 1) if state else None


async def running(total: int, value: int) -> tuple[int, int]:
    await asyncio.sleep(0)
    next_total = total + value
    return next_total, next_total


async def main():
    query = aflow.unfold(3, countdown).map_accum(0, running)
    assert await query.to_list() == [3, 5, 6]

Neither operation starts background work or prefetches the next transition. Cancellation propagates to the active callback and closes the owned iterator chain. Each evaluation begins with the supplied seed by reference; use aflow.defer when mutable state needs a new instance. See the synchronous state examples for tuple ordering and empty-input behavior.

Reuse sliding-window aggregates (2.4.0)

AsyncFlow.window_reduce(size, combine, step=1) uses the same ordered aggregation strategies and algebraic contract as the synchronous operation. The combine function is synchronous. It must be associative and must not modify its arguments; source values and emitted aggregates must remain unchanged during iteration. Awaitable combine functions are not supported.

import asyncio
from operator import add
from fpstreams import aflow

async def main():
    assert await aflow([2, 3, 5, 7]).window_reduce(3, add).to_list() == [10, 15]

Nonoverlapping windows use a running fold without a raw window buffer. The operation creates no background task or prefetch buffer. Cancellation while waiting for the next source item propagates upstream and closes the iterator chain. A stack transfer runs synchronously and can take linear work in the window size; this operation adds no scheduling checkpoint inside a combine call.

Reduce adjacent runs with a Collector (2.4.0)

AsyncFlow.collapse(predicate, collector) follows the same run and Collector rules. The adjacent predicate may be synchronous or awaitable:

import asyncio
from fpstreams import Collectors, aflow

async def adjacent(left, right):
    await asyncio.sleep(0)
    return 0 <= right - left <= 2

async def main():
    result = await aflow([1, 2, 4, 9, 11, 1]).collapse(
        adjacent, Collectors.head(1)
    ).to_list()
    assert result == [[1], [9], [1]]

Collector hooks retain their synchronous protocol. An awaitable returned by the finisher is emitted as data, without an implicit await. No background task or prefetch buffer is added. Cancellation while reading the source or waiting for the predicate closes upstream without finishing the pending run.

Bound pull-ahead work

prefetch(capacity) lets a producer pull ahead while retaining at most capacity accepted values. It preserves encounter order and cancels its owned producer task when downstream stops or fails.

values = await aflow(source).prefetch(32).map_async(transform, concurrency=8).to_list()

Prefetch is useful when upstream latency and downstream work overlap. It is not a concurrency setting for map_async, and increasing the capacity also increases the maximum retained input.

On normal completion, accepted values are emitted in order. If the producer fails, values already accepted by the buffer are emitted before its error. An early downstream stop or external cancellation cleans up the producer, but does not promise delivery of every in-flight value.

With map_async(..., ordered=False), results arrive in completion order. Following it with take(n) need not return the first n inputs, and callbacks whose results are not returned may already have started. With ordered output, a later callback can fail while an earlier result is still pending; a full consumer can receive that error, while take(1) may end after the first result without receiving it. Mapper timeouts use cooperative cancellation: a callback that suppresses cancellation can return a value, and waiting for cleanup can exceed the timeout.

If a task factory rejects a scope's cleanup task, the scope waits for its existing tasks to stop before upstream resources close. The caller that encountered the rejection still receives that submission error. No replacement task is created for this fallback; later close calls can observe that cleanup has finished.

In the 2.5.0 source, cancelling a caller that is already waiting for a cancelled child to stop does not cancel that child again. The wait finishes before caller cancellation propagates, including when a timeout has cancelled an upstream pull. A child that never finishes can delay this wait indefinitely.

switch_map also closes its owned outer iterator if its first pull task cannot be admitted. The mapper is not called when this submission fails. Query ownership also closes an acquired root iterator when an intermediate stage never starts, such as map(...).take(0), without pulling an item. Synchronous iterators adapted by aflow follow the same ownership rule.

Build bounded sessions

session_window(idle_for, max_count=...) groups consecutive values until the source stays quiet for idle_for seconds. max_count is a required hard cap; reaching it flushes the session even if the idle timer has not fired. Source completion flushes the final non-empty session.

Control request rates

Place throttle before concurrent work when an API accepts only a fixed number of requests per time window. The first window may start with a burst; later items wait without filling a background queue.

results = await (
    aflow(requests)
    .throttle(5, per=1.0)  # Start at most five requests per second.
    .map_async(send_request, concurrency=3)  # Keep at most three requests in flight.
    .to_list()
)

Use spaceout(seconds) when every pair of emissions needs a minimum gap. delay(seconds) waits once, before the first upstream item is requested.

In the 2.5.0 source, NaN durations raise ValueError when configuring interval, delay, throttle, spaceout, timeout, debounce, buffer_timeout (also batch_timeout), map_async, and map_batches. session_window already rejects NaN. Zero remains valid for interval and debounce; the other durations must be positive. Positive infinity remains accepted and can leave timed work waiting indefinitely. Validation does not open the source or start a mapper.

Keep only the latest work

switch_map is useful for search boxes, live filters, and other inputs where a newer value makes older work irrelevant. When a new outer item arrives, fpstreams cancels and closes the previous inner source. The last inner source can finish after the outer source ends.

matches = await (
    aflow(query_changes)
    .switch_map(search_pages)  # Switch away from the previous result stream.
    .to_list()
)

search_pages may return an iterable, an async iterable, or an awaitable containing either one. Inner iterators used by merge_map and switch_map do not need an aclose hook. If an inner exposes a callable hook, it is selected when the inner opens and called once during cleanup. Its result is awaited only when it is awaitable. Cancellation and cleanup apply to work owned by the pipeline. Whether an external request, thread, or independently created task actually stops depends on that work's implementation.

In the 2.5.0 source, the concurrency limits of map_async and merge_map, and the result buffer of map_async, must be positive integers. Integer-like values use __index__ and are normalized when configuring the flow, before it opens its source. Floating-point values, including NaN and infinity, raise TypeError.

Reduce without materializing

sum, min, max, and minmax consume the source once without retaining its input items; accumulator size still follows the result type. sum accepts the same start argument and rejects string or bytes starts in the same way as Python's built-in sum.

min, max, and minmax accept a callable, field name, integer index, dotted path, or row expression as key. A callable key may return a value directly or return an awaitable. Equal keys retain the first item encountered.

lowest, highest = await aflow(events).minmax(key="metrics.latency_ms")
total = await aflow(amounts).sum(start=opening_balance)

The extreme-value terminals raise EmptyFlowError on an empty source. sum returns start for an empty source. Exceptions and cancellation close the owned upstream iterator before propagating.

Methods

fpstreams.AsyncFlow

Bases: AsyncFlowTerminalsMixin[T], Generic[T]

An async pipeline that opens its sync or async source when consumption begins.

run_with_report async 

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

Await one eager terminal normally and pair its value with a read-only report.

to_list async 

to_list() -> list[T]

Consume the async flow and collect its items in a list.

Returns:

Type Description
list[T]

All emitted items in encounter order.

to_tuple async 

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

Execute the async pipeline and collect its items in a tuple.

Returns:

Type Description
tuple[T, ...]

All emitted items in encounter order as a tuple.

to_set async 

to_set() -> set[T]

Execute the async pipeline and collect distinct hashable items.

Returns:

Type Description
set[T]

The distinct emitted items; every item must be hashable.

join async 

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.

partition async 

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

Collect matching and non-matching items in separate lists.

Parameters:

Name Type Description Default
predicate Callable[[T], bool | Awaitable[bool]]

Sync or async callable resolved 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 async 

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.

first async 

first() -> T

first(default: R) -> T | R

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

Return the first item and close the async source immediately.

Parameters:

Name Type Description Default
default Any

Returned only when the async 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 async 

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 async 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 async flow is empty and no default is supplied.

find async 

find(predicate: Callable[[T], bool | Awaitable[bool]]) -> T

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

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

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

Parameters:

Name Type Description Default
predicate Callable[[T], bool | Awaitable[bool]]

Sync or async callable resolved in order until its first truthy result.

 required
default Any

Returned when no resolved 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 async 

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

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

Parameters:

Name Type Description Default
predicate Callable[[T], bool | Awaitable[bool]]

Sync or async callable resolved in order until its first truthy result.

 required

Returns:

Type Description
int | None

The zero-based position of the first truthy resolved predicate result, or None.

index_of async 

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 async 

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

count() -> int

Count all items produced by the async flow.

Returns:

Type Description
int

The total number of emitted items.

sum async 

sum(start: Any = 0) -> Any

Add all items to start in one asynchronous traversal.

Parameters:

Name Type Description Default
start Any

Initial value, with the same string and bytes restrictions as Python's built-in sum.

 0

Returns:

Type Description
Any

The total of start and every emitted item.

min async 

min(*, key: Selector | None = None) -> T

Return the first item with the smallest selected value.

Parameters:

Name Type Description Default
key Selector | None

Optional sync or async callable, field name, index, path, or expression used for comparison.

 None

Raises:

Type Description
EmptyFlowError

If the async flow emits no items.

max async 

max(*, key: Selector | None = None) -> T

Return the first item with the largest selected value.

Parameters:

Name Type Description Default
key Selector | None

Optional sync or async callable, field name, index, path, or expression used for comparison.

 None

Raises:

Type Description
EmptyFlowError

If the async flow emits no items.

minmax async 

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

Return the first minimum and maximum items in one traversal.

Parameters:

Name Type Description Default
key Selector | None

Optional sync or async callable, field name, index, path, or expression used for comparison.

 None

Raises:

Type Description
EmptyFlowError

If the async flow emits no items.

collect async 

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

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

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

Reduce the flow with one Collector or named Collectors.

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

Parameters:

Name Type Description Default
collector Callable[[Iterable[T]], C | Awaitable[C]] | Collector[T, Any, C] | None

A streaming Collector, or a sync or async callable that receives a list containing the entire consumed flow.

 None
**collectors Collector[T, Any, Any]

Named streaming collectors updated in one async traversal.

 {}

Returns:

Type Description
C | dict[str, Any]

The collector result, or a dictionary of results for named collectors.

aggregate async 

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

Compute several named aggregations while traversing the async flow once.

Collectors and agg factory results share the same asynchronous traversal.

Parameters:

Name Type Description Default
**aggregations Collector[Any, Any, Any]

Result names mapped to collectors updated in one async traversal.

 {}

Returns:

Type Description
dict[str, Any]

Finished aggregation values keyed by the supplied argument names.

mean async 

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 async 

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 async 

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.

reduce async 

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

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

Reducer return values are resolved once when awaitable. An empty flow returns a supplied initial value unchanged; an unseeded singleton returns its item unchanged, even when that initial value or item is awaitable.

Parameters:

Name Type Description Default
function Callable[[Any, T], Any | Awaitable[Any]]

Sync or async callback invoked 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 async flow is empty and initial is omitted.

reduce_right async 

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

Combine buffered items from right to left.

Reducer return values are resolved once when awaitable. An empty flow returns a supplied initial value unchanged; an unseeded singleton returns its item unchanged, even when that initial value or item is awaitable.

Parameters:

Name Type Description Default
function Callable[[T, Any], Any | Awaitable[Any]]

Sync or async callback invoked 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 async flow is empty and initial is omitted.

reduce_by async 

reduce_by(
    key: Selector,
    function: Callable[[R, T], R | Awaitable[R]],
    *,
    initializer: Callable[[], R | Awaitable[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; callable results may be awaitable.

 required
function Callable[[R, T], R | Awaitable[R]]

Sync or async callback invoked as function(group_state, item).

 required
initializer Callable[[], R | Awaitable[R]]

Sync or async callable invoked when each distinct group is first seen.

 required

Returns:

Type Description
dict[Any, R]

Final resolved accumulator for each hashable key, in first-key encounter order.

frequencies async 

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 resolved selected value.

any async 

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

Return whether at least one item satisfies the predicate.

Parameters:

Name Type Description Default
predicate Callable[[T], bool | Awaitable[bool]]

Sync or async predicate resolved 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 async 

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

Return whether every item satisfies the predicate.

Parameters:

Name Type Description Default
predicate Callable[[T], bool | Awaitable[bool]]

Sync or async predicate resolved 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 async 

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

Return whether no item satisfies predicate.

Parameters:

Name Type Description Default
predicate Callable[[T], bool | Awaitable[bool]]

Sync or async predicate resolved 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.

for_each async 

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

Run an action for every item and return after completion.

Parameters:

Name Type Description Default
action Callable[[T], Any]

Sync or async callable resolved once for every emitted item; return values are ignored.

 required

explain 

explain(
    terminal: AsyncTerminalName = "iterate",
) -> AsyncPlanExplanation

Describe stream facts and terminal completion risks without consuming the source.

Parameters:

Name Type Description Default
terminal AsyncTerminalName

Async terminal whose completion requirements should be analyzed.

 'iterate'

Returns:

Type Description
AsyncPlanExplanation

A lazy explanation view over the current plan and selected terminal.

Raises:

Type Description
ValueError

If terminal is not a supported async terminal name.

of staticmethod 

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

Create an async flow from positional items.

Parameters:

Name Type Description Default
*items R

Positional values to emit in argument order.

 ()

Returns:

Type Description
AsyncFlow[R]

A reusable async flow that emits items in argument order.

from_iterable staticmethod 

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

Create an async flow over a synchronous iterable.

Parameters:

Name Type Description Default
source Iterable[R]

Synchronous iterable adapted to async iteration when consumed.

 required

Returns:

Type Description
AsyncFlow[R]

An async flow that emits each item from source; iterator instances are one-shot.

from_aiterable staticmethod 

from_aiterable(source: AsyncIterable[R]) -> AsyncFlow[R]

Create an async flow over an asynchronous iterable.

Parameters:

Name Type Description Default
source AsyncIterable[R]

Asynchronous iterable opened when consumption begins.

 required

Returns:

Type Description
AsyncFlow[R]

An async flow over source; async iterator instances are one-shot.

unfold staticmethod 

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

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

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

Generate values from sync or async state transitions until None is returned.

Parameters:

Name Type Description Default
seed C

Initial state, reused by reference for each evaluation. Use immutable state or aflow.defer when fresh mutable state is needed.

 required
function Callable[[C], tuple[R, C] | Awaitable[tuple[R, C] | None] | None]

Sync or async callback returning (value, next_state), or None to stop. A tuple's value may be None. Only one callback runs at a time.

 required

Returns:

Type Description
AsyncFlow[R]

A lazy, reopenable async flow that does not emit the seed or request a later transition after early exit. Termination depends on the callback.

Raises:

Type Description
TypeError

If function is not callable.

from_queue staticmethod 

from_queue(
    queue: Queue[R], *, stop: object = _NO_QUEUE_STOP
) -> AsyncFlow[R]

Create a non-owning, one-shot flow over an asyncio queue.

Values are requested lazily and emitted in queue.get() return order. When provided, stop ends the flow by identity and is not emitted. On Python 3.13+, queue shutdown ends the flow normally. Values already removed from the queue are not returned when downstream stops early. This adapter never calls task_done(), including for the hidden stop; do not use pull-ahead prefetch() when relying on Queue.join() or per-item acknowledgements. Queue and producer ownership remain with the caller.

from_file staticmethod 

from_file(
    path: str | PathLike[str], *, encoding: str = "utf-8"
) -> AsyncFlow[str]

Read a text file asynchronously and emit lines without trailing newlines.

Parameters:

Name Type Description Default
path str | PathLike[str]

Text file opened only when the returned flow is consumed.

 required
encoding str

Encoding used by aiofiles.open.

 'utf-8'

Returns:

Type Description
AsyncFlow[str]

A reusable async flow of lines with trailing CR and LF characters removed.

Cancellation waits for an in-flight open, read, or close to finish before releasing the file. A stalled filesystem operation can therefore delay cancellation.

interval staticmethod 

interval(seconds: float) -> AsyncFlow[int]

Emit increasing integers separated by seconds.

Parameters:

Name Type Description Default
seconds float

Delay before the first integer and between later integers.

 required

Returns:

Type Description
AsyncFlow[int]

A reusable, infinite flow emitting 0, 1, 2, ... at the requested interval.

Raises:

Type Description
ValueError

If seconds is negative or NaN.

paginate staticmethod 

paginate(
    fetch_page: Callable[
        [C | None],
        tuple[AsyncIterable[R] | Iterable[R], C | None]
        | Awaitable[
            tuple[AsyncIterable[R] | Iterable[R], C | None]
        ],
    ],
    *,
    start: C | None = None,
) -> AsyncFlow[R]

Fetch pages lazily until the returned cursor is None.

Parameters:

Name Type Description Default
fetch_page Callable[[C | None], tuple[AsyncIterable[R] | Iterable[R], C | None] | Awaitable[tuple[AsyncIterable[R] | Iterable[R], C | None]]]

Sync or async callable receiving the current cursor and returning (page, next_cursor); each page may be sync or async iterable.

 required
start C | None

Cursor passed to the first fetch_page call.

 None

Returns:

Type Description
AsyncFlow[R]

A reusable async flow that flattens each page and stops after a None next cursor.

map_async 

map_async(
    function: Callable[[T], Awaitable[R]],
    *,
    concurrency: int = 8,
    ordered: bool = True,
    timeout: float | None = None,
    buffer: int | None = None,
) -> AsyncFlow[R]

map_async(
    function: Callable[[T], R | Awaitable[R]],
    *,
    concurrency: int = 8,
    ordered: bool = True,
    timeout: float | None = None,
    buffer: int | None = None,
) -> AsyncFlow[R]

map_async(
    function: Callable[[T], R | Awaitable[R]],
    *,
    concurrency: int = 8,
    ordered: bool = True,
    timeout: float | None = None,
    buffer: int | None = None,
) -> AsyncFlow[R]

Map items with bounded concurrency, buffering, ordering, and timeout.

Concurrency is bounded. Ordered mode delays later completed results until earlier inputs finish, while the buffer permits completed work to refill active mapper slots. Callbacks run on the event loop. Timeout and cancellation are cooperative and cannot interrupt a callback that blocks the loop.

When concurrency exceeds one, one additional owned task reads the source so a slow input cannot block completed results. The query-wide task budget includes this reader.

Parameters:

Name Type Description Default
function Callable[[T], R | Awaitable[R]]

Sync or async mapper called once for each source item.

 required
concurrency int

Maximum mapper calls in flight.

 8
ordered bool

Emit in source order when true, otherwise in task-completion order.

 True
timeout float | None

Optional cooperative per-item timeout while awaiting the mapper result.

 None
buffer int | None

Maximum submitted results not yet emitted, or twice concurrency by default.

 None

Returns:

Type Description
AsyncFlow[R]

An async flow of mapped values with bounded work and cleanup on early exit.

Raises:

Type Description
TypeError

If concurrency or buffer is not an integer.
ValueError

If concurrency or buffer is less than one, or timeout is not positive.

map 

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

map(
    function: Callable[[T], R | Awaitable[R]],
) -> AsyncFlow[R]

map(
    function: Callable[[T], R | Awaitable[R]],
) -> AsyncFlow[R]

Apply a synchronous or asynchronous function to each item in order.

map preserves encounter order and awaits awaitable results; work starts only when the flow is consumed.

Parameters:

Name Type Description Default
function Callable[[T], R | Awaitable[R]]

Sync or async mapper called once per item; awaitable results are resolved.

 required

Returns:

Type Description
AsyncFlow[R]

An async flow of mapped values in source order, with one mapper call in flight.

filter 

filter(
    predicate: Callable[[T], bool | Awaitable[bool]],
) -> AsyncFlow[T]

Keep items for which the sync or async predicate is truthy.

The predicate may be synchronous or asynchronous and is evaluated in encounter order.

Parameters:

Name Type Description Default
predicate Callable[[T], bool | Awaitable[bool]]

Sync or async callable resolved for each item; truthy results retain it.

 required

Returns:

Type Description
AsyncFlow[T]

An async flow containing only items whose resolved predicate result is truthy.

where 

where(
    predicate: Callable[[T], bool | Awaitable[bool]],
) -> AsyncFlow[T]

Keep items for which predicate returns a truthy value.

This is an alias-style filtering operation for a synchronous or asynchronous predicate.

Parameters:

Name Type Description Default
predicate Callable[[T], bool | Awaitable[bool]]

Sync or async callable resolved for each item; truthy results retain it.

 required

Returns:

Type Description
AsyncFlow[T]

The same ordered async filtering pipeline produced by filter(predicate).

reject 

reject(
    predicate: Callable[[T], bool | Awaitable[bool]],
) -> AsyncFlow[T]

Drop items for which the sync or async predicate is truthy.

Parameters:

Name Type Description Default
predicate Callable[[T], bool | Awaitable[bool]]

Sync or async callable resolved for each item; truthy results drop it.

 required

Returns:

Type Description
AsyncFlow[T]

An async flow containing only items whose resolved predicate result is falsey.

tap 

tap(action: Callable[[T], Any]) -> AsyncFlow[T]

Run a sync or async side effect while passing each item through.

Parameters:

Name Type Description Default
action Callable[[T], Any]

Sync or async side effect resolved before the original item is emitted.

 required

Returns:

Type Description
AsyncFlow[T]

An async flow that passes every item through unchanged after running action.

flat_map 

flat_map(
    function: Callable[
        [T],
        AsyncIterable[R]
        | Iterable[R]
        | Awaitable[AsyncIterable[R] | Iterable[R]],
    ],
) -> AsyncFlow[R]

Map each item to an iterable and emit its contents.

The mapper may be synchronous or asynchronous; each returned iterable is flattened in encounter order.

Parameters:

Name Type Description Default
function Callable[[T], AsyncIterable[R] | Iterable[R] | Awaitable[AsyncIterable[R] | Iterable[R]]]

Sync or async mapper returning a sync or async iterable for each item.

 required

Returns:

Type Description
AsyncFlow[R]

An async flow that drains each mapped iterable in source order before mapping the next item.

merge 

merge(
    *others: AsyncIterable[T] | Iterable[T],
) -> AsyncFlow[T]

Merge this flow with other sources in completion order.

Parameters:

Name Type Description Default
*others AsyncIterable[T] | Iterable[T]

Sync or async sources to interleave with this flow.

 ()

Returns:

Type Description
AsyncFlow[T]

An async flow that emits each source's next item as its pull completes.

combine_latest 

combine_latest(
    *others: AsyncIterable[Any] | Iterable[Any],
) -> AsyncFlow[tuple[Any, ...]]

Emit the latest value from every source after all have produced once.

Parameters:

Name Type Description Default
*others AsyncIterable[Any] | Iterable[Any]

Sync or async sources whose latest values join this flow's latest value.

 ()

Returns:

Type Description
AsyncFlow[tuple[Any, ...]]

A flow of latest-value tuples in source argument order. It starts emitting after every source has produced a value. Completed sources keep their final value in later tuples.

merge_map 

merge_map(
    function: Callable[
        [T],
        AsyncIterable[R]
        | Iterable[R]
        | Awaitable[AsyncIterable[R] | Iterable[R]],
    ],
    *,
    concurrency: int = 8,
) -> AsyncFlow[R]

Map items to inner sources and merge them with bounded concurrency.

Parameters:

Name Type Description Default
function Callable[[T], AsyncIterable[R] | Iterable[R] | Awaitable[AsyncIterable[R] | Iterable[R]]]

Sync or async mapper returning a sync or async inner source.

 required
concurrency int

Shared maximum for inner sources being opened or actively consumed.

 8

Returns:

Type Description
AsyncFlow[R]

An async flow interleaving inner items in completion order under the concurrency cap.

Raises:

Type Description
TypeError

If function is not callable or concurrency is not an integer.
ValueError

If concurrency is less than one.

switch_map 

switch_map(
    function: Callable[
        [T],
        AsyncIterable[R]
        | Iterable[R]
        | Awaitable[AsyncIterable[R] | Iterable[R]],
    ],
) -> AsyncFlow[R]

Map each item to a source and emit only from the latest source.

When a new outer item arrives, the previous mapper or inner source is cancelled and closed. After the outer source completes, the latest inner source may finish normally.

Parameters:

Name Type Description Default
function Callable[[T], AsyncIterable[R] | Iterable[R] | Awaitable[AsyncIterable[R] | Iterable[R]]]

A sync or async callable returning a sync or async iterable.

 required

Returns:

Type Description
AsyncFlow[R]

An async flow that emits only from the most recently mapped inner source.

Raises:

Type Description
TypeError

If function is not callable.

delay 

delay(seconds: float) -> AsyncFlow[T]

Wait before requesting the first item from this flow.

The delay is applied once per evaluation. No upstream item is requested while waiting, and cancellation during the delay still closes the source.

Parameters:

Name Type Description Default
seconds float

The positive delay in seconds before the first upstream request.

 required

Returns:

Type Description
AsyncFlow[T]

An async flow that waits once before its first upstream pull, then forwards normally.

Raises:

Type Description
ValueError

If seconds is not positive.

throttle 

throttle(max_count: int, *, per: float) -> AsyncFlow[T]

Limit emissions within a sliding time window.

Up to max_count items may be emitted immediately. Later items wait until the oldest emission leaves the monotonic window; encounter order and pull-based backpressure are preserved.

Parameters:

Name Type Description Default
max_count int

Maximum emissions permitted in each rolling per-second window.

 required
per float

The positive sliding-window duration in seconds.

 required

Returns:

Type Description
AsyncFlow[T]

An async flow that delays, but never drops, items to enforce the rolling rate limit.

Raises:

Type Description
TypeError

If max_count is not an integer.
ValueError

If max_count is below one or per is not positive.

spaceout 

spaceout(seconds: float) -> AsyncFlow[T]

Separate consecutive emissions by at least seconds.

The first item is emitted immediately. Each later item waits only for the remainder of the requested interval, using the event loop's monotonic clock.

Parameters:

Name Type Description Default
seconds float

The positive minimum interval between emissions.

 required

Returns:

Type Description
AsyncFlow[T]

An async flow that emits the first item immediately and delays later items as needed.

Raises:

Type Description
ValueError

If seconds is not positive.

timeout 

timeout(seconds: float) -> AsyncFlow[T]

Fail when the next item takes longer than seconds.

Parameters:

Name Type Description Default
seconds float

Maximum wait for each individual upstream anext call.

 required

Returns:

Type Description
AsyncFlow[T]

An async flow that raises TimeoutError and cancels an overdue item pull.

Raises:

Type Description
ValueError

If seconds is not positive.

debounce 

debounce(seconds: float) -> AsyncFlow[T]

Emit an item only after the source stays quiet for seconds.

Parameters:

Name Type Description Default
seconds float

Quiet interval required before the latest pending item is emitted.

 required

Returns:

Type Description
AsyncFlow[T]

An async flow that replaces a pending item whenever a newer one arrives. The final pending item is emitted when the source completes.

Raises:

Type Description
ValueError

If seconds is negative or NaN.

buffer_timeout 

buffer_timeout(
    max_count: int, seconds: float
) -> AsyncFlow[tuple[T, ...]]

Flush a tuple when it reaches max_count or seconds elapse.

Parameters:

Name Type Description Default
max_count int

Item count that flushes the current batch immediately.

 required
seconds float

Maximum time from a batch's first item until that batch is flushed.

 required

Returns:

Type Description
AsyncFlow[tuple[T, ...]]

An async flow of non-empty tuples flushed by count, timeout, or source completion.

session_window 

session_window(
    idle_for: float, *, max_count: int
) -> AsyncFlow[tuple[T, ...]]

Group consecutive items until the source stays idle or the hard count cap is reached.

The processing-time idle timer is reset after every accepted item. Source completion flushes the final non-empty tuple.

Parameters:

Name Type Description Default
idle_for float

Positive quiet interval that closes the current session.

 required
max_count int

Required maximum number of items retained in one session.

 required

Returns:

Type Description
AsyncFlow[tuple[T, ...]]

An async flow of non-empty session tuples in encounter order.

prefetch 

prefetch(capacity: int) -> AsyncFlow[T]

Pull upstream values ahead under an explicit bounded buffer.

Parameters:

Name Type Description Default
capacity int

Maximum accepted upstream values not yet handed to downstream.

 required

Returns:

Type Description
AsyncFlow[T]

An async flow preserving every value in encounter order.

Raises:

Type Description
TypeError

If capacity does not implement the integer index protocol.
ValueError

If capacity is less than one.

filter_map 

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

Map items and discard results that are None.

Parameters:

Name Type Description Default
function Callable[[T], R | Awaitable[R | None] | None]

Sync or async mapper returning one output value or None.

 required

Returns:

Type Description
AsyncFlow[R]

An async flow of non-None mapped results; falsey values such as 0 are retained.

pluck 

pluck(selector: Selector) -> AsyncFlow[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
AsyncFlow[Any]

An async flow containing the value selected from each source item.

compact 

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

Drop items whose own or selected value is falsey.

Parameters:

Name Type Description Default
selector Selector | None

Optional callable, field name, index, path, or expression whose truth value determines whether the original item is retained.

 None

Returns:

Type Description
AsyncFlow[T]

An async flow keeping items whose own or selected value is truthy. Without a selector, this drops None, 0, False, and empty containers.

filter_none 

filter_none() -> AsyncFlow[T]

Drop only items equal to None, retaining every other falsey value.

Returns:

Type Description
AsyncFlow[T]

An async flow containing every non-None source item.

unique 

unique() -> AsyncFlow[T]

Keep the first occurrence of each value in source order.

Returns:

Type Description
AsyncFlow[T]

An async flow keeping the first occurrence of each distinct value. Unhashable values are compared by equality.

unique_by 

unique_by(selector: Selector) -> AsyncFlow[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; callable results may be awaitable.

 required

Returns:

Type Description
AsyncFlow[T]

An async flow containing the first item for each distinct resolved selected key.

attempt 

attempt(
    function: Callable[[T], R | Awaitable[R]],
) -> AsyncFlow[Result[R]]

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

Parameters:

Name Type Description Default
function Callable[[T], R | Awaitable[R]]

Sync or async mapper that may raise an Exception.

 required

Returns:

Type Description
AsyncFlow[Result[R]]

An async flow of Ok mapped values and Err objects for raised exceptions.

take 

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

Emit at most count items and cancel pending upstream work.

Parameters:

Name Type Description Default
count SupportsIndex

Maximum number of leading items to emit.

 required

Returns:

Type Description
AsyncFlow[T]

An async flow of at most count items. Reaching the limit cancels pending upstream work and closes its iterators.

Raises:

Type Description
TypeError

If count does not support the integer index protocol.
ValueError

If count is negative.

drop 

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

Skip count items before yielding the remainder.

Parameters:

Name Type Description Default
count SupportsIndex

Number of leading source items to consume without emitting.

 required

Returns:

Type Description
AsyncFlow[T]

An async 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.

take_while 

take_while(
    predicate: Callable[[T], bool | Awaitable[bool]],
) -> AsyncFlow[T]

Emit the longest prefix that satisfies predicate.

Parameters:

Name Type Description Default
predicate Callable[[T], bool | Awaitable[bool]]

Sync or async callable resolved on leading items until its first falsey result.

 required

Returns:

Type Description
AsyncFlow[T]

An async flow ending before the first item whose resolved predicate is falsey.

take_while_inclusive 

take_while_inclusive(
    predicate: Callable[[T], bool | Awaitable[bool]],
) -> AsyncFlow[T]

Emit through the first item that fails predicate.

Parameters:

Name Type Description Default
predicate Callable[[T], bool | Awaitable[bool]]

Sync or async callable resolved on leading items through its first falsey result.

 required

Returns:

Type Description
AsyncFlow[T]

An async flow ending after emitting the first item whose resolved predicate is falsey.

drop_while 

drop_while(
    predicate: Callable[[T], bool | Awaitable[bool]],
) -> AsyncFlow[T]

Skip the longest prefix that satisfies predicate.

Parameters:

Name Type Description Default
predicate Callable[[T], bool | Awaitable[bool]]

Sync or async callable resolved only while leading items are truthy.

 required

Returns:

Type Description
AsyncFlow[T]

An async flow starting with the first item for which predicate is falsey. Remaining items pass through without further predicate calls.

chunk 

chunk(
    size: int, *, strict: bool = False
) -> AsyncFlow[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
AsyncFlow[tuple[T, ...]]

An async 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, ...]], Awaitable[R]],
    *,
    size: int,
    concurrency: int = 1,
    ordered: bool = True,
    timeout: float | None = None,
    buffer: int | None = None,
) -> AsyncFlow[R]

map_batches(
    function: Callable[[tuple[T, ...]], R | Awaitable[R]],
    *,
    size: int,
    concurrency: int = 1,
    ordered: bool = True,
    timeout: float | None = None,
    buffer: int | None = None,
) -> AsyncFlow[R]

map_batches(
    function: Callable[[tuple[T, ...]], R | Awaitable[R]],
    *,
    size: int,
    concurrency: int = 1,
    ordered: bool = True,
    timeout: float | None = None,
    buffer: int | None = None,
) -> AsyncFlow[R]

Apply a sync or async function to consecutive tuples, one result per batch.

Batches are lazy and include a shorter final batch. Empty input does not call the function. Results stay intact; use flat_map(iter) to expand returned iterables. Awaitable results are resolved before emission. Early exit cancels pending callbacks and closes the source through the existing async map scheduler. Callbacks run on the event loop: timeout and cancellation are cooperative and cannot interrupt a callback that blocks the loop.

Parameters:

Name Type Description Default
function Callable[[tuple[T, ...]], R | Awaitable[R]]

Receives up to size consecutive items and returns a result or awaitable.

 required
size int

Maximum input items per batch.

 required
concurrency int

Maximum batch callbacks in flight; one by default.

 1
ordered bool

Emit in batch order when true, otherwise in completion order.

 True
timeout float | None

Optional cooperative per-batch timeout while awaiting the callback.

 None
buffer int | None

Maximum submitted batches not yet emitted; twice concurrency by default.

 None

Returns:

Type Description
AsyncFlow[R]

A lazy async flow containing one resolved result per batch.

Raises:

Type Description
TypeError

If size does not support the integer index protocol.
ValueError

If size, concurrency, buffer, or timeout is not positive.

window 

window(
    size: int, *, step: int = 1
) -> AsyncFlow[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
AsyncFlow[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,
) -> AsyncFlow[T]

Reduce sliding windows using an associative synchronous function.

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. Awaitable callbacks are not supported.

 required
step SupportsIndex

Number of source items between successive window starts.

 1

Returns:

Type Description
AsyncFlow[T]

An async flow of ordered window aggregates. Empty input emits nothing; initially short input emits one partial result, and trailing partial windows after a full window 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.
Notes

Uses the same algebraic contract and cached state as Flow.window_reduce. Callback grouping and exception timing can differ from window().map_async(...). Floating-point addition is not associative. Cached entry count is bounded by size + 1, not by bytes. Combine calls are amortized constant per item, with linear stack transfers. No background task or pull-ahead buffer is introduced.

pairwise 

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

Emit each adjacent pair of items.

Returns:

Type Description
AsyncFlow[tuple[T, T]]

An async flow of overlapping (previous, current) pairs. Fewer than two items produce no pairs.

pair_map 

pair_map(
    function: Callable[[T, T], R | Awaitable[R]],
) -> AsyncFlow[R]

Apply a two-argument function to each adjacent pair.

Parameters:

Name Type Description Default
function Callable[[T, T], R | Awaitable[R]]

Sync or async mapper called as function(previous, current).

 required

Returns:

Type Description
AsyncFlow[R]

An async flow containing one resolved mapped result per adjacent pair.

group_runs 

group_runs(
    key: Selector | None = None,
) -> AsyncFlow[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. Callable results may be awaitable.

 None

Returns:

Type Description
AsyncFlow[tuple[T, ...]]

An async flow of non-empty tuples, one for each contiguous run of equal resolved keys.

enumerate 

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

Pair each item with a consecutive index starting at start.

Parameters:

Name Type Description Default
start int

Integer-compatible index paired with the first source item.

 0

Returns:

Type Description
AsyncFlow[tuple[int, T]]

An async flow of (index, item) pairs with consecutive integer indices.

zip 

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

Pair items with another source until one side ends.

Parameters:

Name Type Description Default
other AsyncIterable[U] | Iterable[U]

Sync or async source providing the right-hand item in each pair.

 required
strict bool

Raise ValueError during consumption when the two sources have different lengths.

 False

Returns:

Type Description
AsyncFlow[tuple[T, U]]

An async flow of pairs ending with the shorter source unless strict is true.

zip_longest 

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

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

Parameters:

Name Type Description Default
other AsyncIterable[U] | Iterable[U]

Sync or async source providing right-hand values.

 required
fillvalue Any

Substitute used on whichever source is exhausted first.

 None

Returns:

Type Description
AsyncFlow[tuple[T | Any, U | Any]]

An async flow of pairs whose length matches the longer source.

intersperse 

intersperse(separator: T) -> AsyncFlow[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
AsyncFlow[T]

An async flow with separator between source items and never at either boundary.

concat 

concat(
    *others: AsyncIterable[T] | Iterable[T],
) -> AsyncFlow[T]

Emit this flow followed by each supplied source.

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 AsyncIterable[T] | Iterable[T]

Sync or async sources opened and drained after this flow, in argument order.

 ()

Returns:

Type Description
AsyncFlow[T]

An async flow that drains this source and then each additional source in order.

cross 

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

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

Parameters:

Name Type Description Default
other AsyncIterable[U] | Iterable[U]

Sync or async source buffered after the first left item arrives.

 required
max_right int | None

Optional maximum number of right-side items that may be buffered.

 None

Returns:

Type Description
AsyncFlow[tuple[T, U]]

An async flow of (left, right) pairs. For each left item, it emits a pair with every buffered right item in order.

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,
) -> AsyncFlow[tuple[int, float | None, float | None]]

Emit count, mean, and variance for each recent count-based window.

Parameters:

Name Type Description Default
size SupportsIndex

Positive integer window width in source items, including missing values.

 required
selector Selector | None

Synchronous numeric selector; None selects each whole item. Selected None occupies a window slot but is excluded from the statistics.

 None
ddof SupportsIndex

Nonnegative integer subtracted from valid count when computing variance.

 0

Returns:

Type Description
AsyncFlow[tuple[int, float | None, float | None]]

An async flow of (count, mean, variance) tuples, including warmup windows. Undefined statistics are None. Each evaluation owns fresh state, retains at most size converted values, and uses the same shifted updates and periodic rebuilding as Flow.rolling_statistics. Selectors stay synchronous; nonfinite numeric input is rejected.

scan 

scan(
    initial: R, function: Callable[[R, T], R | Awaitable[R]]
) -> AsyncFlow[R]

Emit each resolved 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 | Awaitable[R]]

Sync or async callback invoked as function(state, item).

 required

Returns:

Type Description
AsyncFlow[R]

An async flow with one accumulated state for every source item.

map_accum 

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

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

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

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

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

Map items sequentially while threading state through sync or async callbacks.

Parameters:

Name Type Description Default
initial C

State for the first callback, reused by reference for each evaluation. Use aflow.defer if mutable state must be freshly allocated.

 required
function Callable[[C, T], tuple[C, R | Awaitable[R]] | Awaitable[tuple[C, R | Awaitable[R]]]]

Called as function(state, item) and resolves to (next_state, output). Each transition and its output are awaited once if awaitable, before the next item is requested. The next state itself is not awaited.

 required

Returns:

Type Description
AsyncFlow[R] | AsyncFlow[Awaitable[R]]

A lazy async flow with one output per item, including None outputs. Empty input emits nothing, and the final state is not emitted. Each evaluation starts from initial, following scan semantics.

Raises:

Type Description
TypeError

If function is not callable.

scan_right 

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

Buffer the source, fold right, and emit resolved 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 | Awaitable[R]]

Sync or async callback invoked as function(item, state) from right to left.

 required
max_items int | None

Optional maximum number of source items that may be buffered.

 None

Returns:

Type Description
AsyncFlow[R]

An async flow with one right-fold state per source item, ordered like the source.

Raises:

Type Description
BufferLimitError

During consumption if the source exceeds max_items.

prepend 

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

Emit values before the items from this flow.

Parameters:

Name Type Description Default
*values T

Items emitted before the first source item, in argument order.

 ()

Returns:

Type Description
AsyncFlow[T]

An async flow containing values followed by every source item.

append 

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

Emit values after the items from this flow.

Parameters:

Name Type Description Default
*values T

Items emitted after the source completes, in argument order.

 ()

Returns:

Type Description
AsyncFlow[T]

An async flow containing every source item followed by values.

map_first 

map_first(
    function: Callable[[T], T | Awaitable[T]],
) -> AsyncFlow[T]

Transform only the first item, if one exists.

Parameters:

Name Type Description Default
function Callable[[T], T | Awaitable[T]]

Sync or async mapper for the first item; never called for an empty source.

 required

Returns:

Type Description
AsyncFlow[T]

An async flow with only its first item replaced by the resolved mapped value.

map_last 

map_last(
    function: Callable[[T], T | Awaitable[T]],
) -> AsyncFlow[T]

Transform only the last item, if one exists.

Parameters:

Name Type Description Default
function Callable[[T], T | Awaitable[T]]

Sync or async mapper for the final item; never called for an empty source.

 required

Returns:

Type Description
AsyncFlow[T]

An async flow with only its final item replaced by the resolved mapped value.

collapse 

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

collapse(
    collapsible: Callable[[T, T], bool | Awaitable[bool]],
    merger: Callable[[T, T], T | Awaitable[T]],
) -> AsyncFlow[T]

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

Merge adjacent items while collapsible returns true.

Parameters:

Name Type Description Default
collapsible Callable[[T, T], bool | Awaitable[bool]]

Sync or async predicate called on neighboring original items.

 required
merger Callable[[T, T], T | Awaitable[T]] | Collector[T, C, R]

Sync or async binary callback, or a Collector whose lifecycle hooks follow the synchronous Collector protocol.

 required

Returns:

Type Description
AsyncFlow[T] | AsyncFlow[R]

An async flow containing one aggregate or finished result per contiguous run. Empty input emits nothing and does not initialize a collector.
Notes

Collector initialization, completion and boundary behavior follow Flow.collapse. The adjacent predicate may be awaitable. Collector hooks are synchronous; a finisher returning an awaitable emits that object without awaiting it. No task or prefetch buffer is added. Upstream failure or cancellation does not finish the current run.

fold 

fold(
    initializer: Callable[[], R | Awaitable[R]],
    function: Callable[[R, T], R | Awaitable[R]],
) -> AsyncFlow[R]

Reduce the whole source and emit one resolved accumulator.

Parameters:

Name Type Description Default
initializer Callable[[], R | Awaitable[R]]

Sync or async callable invoked once per evaluation for initial state.

 required
function Callable[[R, T], R | Awaitable[R]]

Sync or async callback invoked as function(state, item).

 required

Returns:

Type Description
AsyncFlow[R]

An async flow containing one final state. For an empty source, that state is the resolved result of initializer().

fold_async 

fold_async(
    initializer: Callable[[], Awaitable[R]],
    function: Callable[[R, T], Awaitable[R]],
) -> AsyncFlow[R]

Fold with explicitly asynchronous hooks and a resolved state type.

Parameters:

Name Type Description Default
initializer Callable[[], Awaitable[R]]

Called once per evaluation and awaited once to create state.

 required
function Callable[[R, T], Awaitable[R]]

Called for every item and awaited once to replace state.

 required

Returns:

Type Description
AsyncFlow[R]

A lazy async flow emitting exactly one final state, including on empty input. If the resolved state is itself awaitable, it is retained unchanged.
Notes

Annotate the resulting AsyncFlow[R] to declare the accumulator contract and provide context for a lambda step. Hooks must return awaitables. As with other typed callbacks, Any can bypass this static contract; no runtime state validation is added.

batch_by_size 

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

Build batches constrained by count and measured size.

Parameters:

Name Type Description Default
max_size int

Maximum sum of resolved item sizes in a normal batch.

 required
max_count int | None

Optional maximum item count per batch.

 None
get_size Callable[[T], int | Awaitable[int]]

Sync or async callable returning a non-negative integer size; 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
AsyncFlow[tuple[T, ...]]

An async 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.