The asynctools library

The asyncstdlib.asynctools library implements the core toolset used by asyncstdlib itself and similar utilities. All documented members of this module are separate from internal implementation and stable regardless of asyncstdlib internals.

New in version 1.1.0.

Iterator lifetime

borrow(iterator: async iter T) async iter T[source]

Borrow an async iterator, preventing to aclose it

When borrowing an async iterator, the original owner assures to close the iterator as needed. In turn, the borrowed iterator does not allow closing the underlying iterator.

The borrowed iterator supports asend() and athrow() if the underlying iterator supports them as well; this allows borrowing either an AsyncIterator or AsyncGenerator. Regardless of iterator type, aclose() is always provided; it closes only the borrowed iterator, not the underlying iterator.

See also

Use scoped_iter() to ensure an (async) iterable is eventually closed and only borrowed until then.

async with scoped_iter(iterable: (async) iter T) as :async iter T[source]

Context manager that provides an async iterator for an (async) iterable

Roughly equivalent to combining iter() with closing. The resulting asynchronous iterator is automatically borrowed to prevent premature closing when passing the iterator around.

from collections import deque
import asyncstdlib as a

async def head_tail(iterable, leading=5, trailing=5):
    '''Provide the first ``leading`` and last ``trailing`` items'''
    # create async iterator valid for the entire block
    async with scoped_iter(iterable) as async_iter:
        # ... safely pass it on without it being closed ...
        async for item in a.isclice(async_iter, leading):
            yield item
        tail = deque(maxlen=trailing)
        # ... and use it again in the block
        async for item in async_iter:
    for item in tail:
        yield item

Nested scoping of the same iterator is safe: inner scopes automatically forfeit closing the underlying iterator in favour of the outermost scope. This allows passing the scoped iterator to other functions that use scoped_iter().

Async transforming

async for :T in await_each(awaitables: iter await T)[source]

Iterate through awaitables and await each item

This converts an iterable of async into an async iterator of awaited values. Consequently, we can apply various functions made for AsyncIterable[T] to Iterable[Awaitable[T]] as well.


import asyncstdlib as a

 async def check1() -> bool:

async def check2() -> bool:

async def check3() -> bool:

 okay = await a.all(
         [check1(), check2(), check3()]))

New in version 3.9.1.

await apply(func: (*T, **T) -> R, *args: await T, **kwargs: await T) R[source]

Await the arguments and keyword arguments and then apply func on them


async def compute_something() -> float:

async def compute_something_else() -> float:

result = await apply(
    lambda x, y: x ** y,

The function apply serves, for example, a practical use case when you want to chain operations on awaitables and need to pass around the final awaitable for further operations.

New in version 3.9.1.