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.

Added in version 1.1.0.

Iterator lifetime

borrow(iterator: async iter T) → async iter T

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

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 a.scoped_iter(iterable) as async_iter:
        # ... safely pass it on without it being closed ...
        async for item in a.islice(async_iter, leading):
            yield item
        tail = deque(maxlen=trailing)
        # ... and use it again in the block
        async for item in async_iter:
            tail.append(item)
    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

sync(function: (...) -> (await) T) -> (...) → await T

Wraps a callable to ensure its result can be awaited

Useful to write async neutral functions by wrapping callable arguments, or to use synchronous functions where asynchronous ones are expected. Wrapping a regular function defined using def or lambda makes it behave roughly as if it were defined using async def instead.

Example:

import asyncstdlib as a

def test1_sync(x, y):
    ...

async def test1_async(x):
    ...

async def main():
    await a.sync(test1_sync)(x=1, y=2)
    await a.sync(test1_async)(x=8)
    await a.sync(lambda x: x ** 3)(x=5)

if __name__ == "__main__":
    asyncio.run(main())

Note

This should never be applied as the sole decorator on a function. Define the function as async def instead.

Added in version 3.9.3.

async for :T in any_iter(iter: (await) (async) iter (await) T)

Provide an async iterator for various forms of “asynchronous iterable”

Useful to uniformly handle async iterables, awaitable iterables, iterables of awaitables, and similar in an async for loop. Among other things, this matches all forms of async def functions providing iterables.

import random
import asyncstdlib as a

# AsyncIterator[T]
async def async_iter(n):
    for i in range(n):
        yield i

# Awaitable[Iterator[T]]
async def await_iter(n):
    return [*range(n)]

some_iter = random.choice([async_iter, await_iter, range])
async for item in a.any_iter(some_iter(4)):
    print(item)

This function must eagerly resolve each “async layer” before checking if the next layer is as expected. This incurs a performance penalty and non-iterables may be left unusable by this. Prefer iter() to test for iterables with EAFP and for performance when only simple iterables need handling.

Added in version 3.10.3.

async for :T in await_each(awaitables: iter await T)

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.

Example:

import asyncstdlib as a

async def check1() -> bool:
    ...

async def check2() -> bool:
    ...

async def check3() -> bool:
    ...

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

Added in version 3.9.1.

await apply(func: (*T, **T) -> R, *args: await T, **kwargs: await T) → R

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

Example:

async def compute_something() -> float:
    ...

async def compute_something_else() -> float:
    ...

result = await apply(
    lambda x, y: x ** y,
    compute_something(),
    compute_something_else())

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.

Added in version 3.9.1.