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.

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

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.

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.

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)  # Cancel the previous search when the query changes.
    .to_list()
)

search_pages may return an iterable, an async iterable, or an awaitable containing either one.

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

Compute several named aggregations while traversing the async flow once.

All named aggregators are updated during the same asynchronous traversal.

Parameters:

Name	Type	Description	Default
**aggregations	Aggregator	Result names mapped to aggregators 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.

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.

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.

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.

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.

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

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 per-item deadline covering mapper invocation and awaiting its 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
ValueError	If concurrency or buffer is less than one.

map

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.

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.

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.

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 equal to 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: int) -> AsyncFlow[T]

Emit at most count items and cancel pending upstream work.

Parameters:

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

drop

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

Skip count items before yielding the remainder.

Parameters:

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

Returns:

Type	Description
AsyncFlow[tuple[T, ...]]	An async flow of non-overlapping tuples, including a final shorter tuple when needed.

Raises:

Type	Description
ValueError	If size is less than one.

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

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.

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.

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.

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

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]]	Sync or async callback combining the current run aggregate with the next item.	required

Returns:

Type	Description
AsyncFlow[T]	An async flow containing one resolved aggregate per contiguous collapsible 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().

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.