The asynctools library¶
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
New in version 1.1.0.
- borrow(iterator: async iter T) async iter T ¶
Borrow an async iterator, preventing to
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
athrow()if the underlying iterator supports them as well; this allows borrowing either an
AsyncGenerator. Regardless of iterator type,
aclose()is always provided; it closes only the borrowed iterator, not the underlying iterator.
- async with scoped_iter(iterable: (async) iter T) as :async iter T¶
Context manager that provides an async iterator for an (async)
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: 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
- sync(function: (...) -> (await) T) -> (...) await T ¶
Wraps a callable to ensure its result can be
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
lambdamakes it behave roughly as if it were defined using
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())
New in version 3.9.3.
- async for :T in await_each(awaitables: iter await T)¶
awaitablesand await each item
This converts an iterable of async into an async iterator of awaited values. Consequently, we can apply various functions made for
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()]) )
New 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
async def compute_something() -> float: ... async def compute_something_else() -> float: ... result = await apply( lambda x, y: x ** y, compute_something(), compute_something_else())
applyserves, 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.