or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

docs

asynctools.mdbuiltins.mdcontextlib.mdfunctools.mdheapq.mdindex.mditertools.md
tile.json

tessl/pypi-asyncstdlib

The missing async toolbox - re-implements functions and classes of the Python standard library to make them compatible with async callables, iterables and context managers

Workspace
tessl
Visibility
Public
Created
Last updated
Describes
pypipkg:pypi/asyncstdlib@3.13.x

To install, run

npx @tessl/cli install tessl/pypi-asyncstdlib@3.13.0

index.mddocs/

asyncstdlib

The missing async toolbox that provides async-compatible versions of Python standard library functions and classes. It offers async implementations for builtins, itertools, functools, contextlib, and heapq modules, plus custom utilities for working with async iterables safely.

Package Information

  • Package Name: asyncstdlib
  • Language: Python
  • Installation: pip install asyncstdlib
  • Python Support: 3.8+
  • Dependencies: Zero runtime dependencies

Core Imports

import asyncstdlib

Most commonly, import specific functions directly:

from asyncstdlib import zip, map, filter, enumerate, all, any
from asyncstdlib import reduce, lru_cache
from asyncstdlib import contextmanager, closing
from asyncstdlib import chain, accumulate
from asyncstdlib import borrow, scoped_iter

Basic Usage

import asyncstdlib

async def example():
    # Async versions of built-in functions
    async_iter = asyncstdlib.map(lambda x: x * 2, [1, 2, 3, 4, 5])
    doubled_values = await asyncstdlib.list(async_iter)
    print(doubled_values)  # [2, 4, 6, 8, 10]
    
    # Works with any async iterable
    async def async_generator():
        for i in range(5):
            yield i
    
    result = await asyncstdlib.sum(async_generator())
    print(result)  # 10
    
    # Safe async iteration
    async with asyncstdlib.scoped_iter(async_generator()) as scoped:
        async for value in scoped:
            print(value)
            if value >= 2:
                break
        # Automatically cleaned up, no StopAsyncIteration leaks

# Run the example
import asyncio
asyncio.run(example())

Architecture

asyncstdlib is designed around the principle of async neutrality - it works seamlessly with any async event loop (asyncio, trio, or custom implementations) without favoring specific frameworks. The library maintains the exact same API signatures as the standard library counterparts while providing async-compatible behavior.

Key design principles:

  • Drop-in replacements: Same function signatures as standard library equivalents
  • Safe iteration: Proper cleanup of async iterators to prevent resource leaks
  • Type safety: Complete type annotations and stub files for full typing support
  • Performance: Optimized implementations designed specifically for async contexts

Capabilities

Built-in Functions

Async versions of Python's built-in functions for iteration, aggregation, and data processing. These provide the fundamental async building blocks that mirror Python's standard builtins.

async def anext(iterator, default=...): ...
def zip(*iterables, strict=False): ...
def map(function, *iterables): ...
def filter(function, iterable): ...
def enumerate(iterable, start=0): ...
def iter(subject, sentinel=None): ...
async def all(iterable): ...
async def any(iterable): ...
async def max(iterable, key=None, default=...): ...
async def min(iterable, key=None, default=...): ...
async def sum(iterable, start=0): ...
async def list(iterable=()): ...
async def dict(iterable=(), **kwargs): ...
async def set(iterable=()): ...
async def tuple(iterable=()): ...
async def sorted(iterable, key=None, reverse=False): ...

Built-in Functions

Iteration Tools

Async versions of itertools functions providing powerful iteration patterns including infinite iterators, filtering, grouping, and combinatorial functions.

def accumulate(iterable, function=None, initial=None): ...
def batched(iterable, n, strict=False): ...
def cycle(iterable): ...
def chain(*iterables): ...
def compress(data, selectors): ...
def dropwhile(predicate, iterable): ...
def filterfalse(predicate, iterable): ...
def islice(iterable, start, stop=None, step=None): ...
def takewhile(predicate, iterable): ...
def starmap(function, iterable): ...
def tee(iterable, n=2, lock=None): ...
def pairwise(iterable): ...
def zip_longest(*iterables, fillvalue=None): ...
def groupby(iterable, key=None): ...

Iteration Tools

Functional Programming

Async versions of functools utilities including reduction, caching, and property decorators designed for async functions and methods.

async def reduce(function, iterable, initial=None): ...
def lru_cache(maxsize=128, typed=False): ...
def cache(user_function): ...
def cached_property(getter): ...

Functional Programming

Context Management

Async context managers and decorators for resource management, providing async versions of contextlib utilities with additional safety features.

def contextmanager(func): ...
class ContextDecorator: ...
class closing: ...
class nullcontext: ...
class ExitStack: ...

Context Management

Heap Operations

Async versions of heapq functions for working with priority queues and sorted data structures in async contexts.

def merge(*iterables, key=None, reverse=False): ...
async def nlargest(iterable, n, key=None): ...
async def nsmallest(iterable, n, key=None): ...

Heap Operations

Async Utilities

Custom utilities specific to asyncstdlib for safely working with async iterables, borrowing iterators, scoped iteration, and bridging sync/async contexts.

def borrow(iterable): ...
def scoped_iter(iterable): ...
async def await_each(awaitables): ...
def any_iter(*iterables): ...
async def apply(function, *args, **kwargs): ...
def sync(async_function, *args, **kwargs): ...

Async Utilities

Types

Core types used throughout the asyncstdlib API:

# Type variables
T = TypeVar('T')                    # Generic type
T1 = TypeVar('T1')                  # Generic type variants
T2 = TypeVar('T2')
T3 = TypeVar('T3')
T4 = TypeVar('T4')
T5 = TypeVar('T5')
R = TypeVar('R')                    # Return type
LT = TypeVar('LT')                  # Less-than comparable type
HK = TypeVar('HK', bound=Hashable)  # Hashable key type
ADD = TypeVar('ADD')                # Addable type
S = TypeVar('S')                    # Send type for generators

# Type aliases for any iterable (sync or async)
AnyIterable = Union[Iterable[T], AsyncIterable[T]]

# Async iterator that can be closed
class AsyncIterator(Protocol[T]):
    async def __anext__(self) -> T: ...
    async def aclose(self) -> None: ...

# Async generator protocol
class AsyncGenerator(AsyncIterator[T], Generic[T, S]):
    async def asend(self, value: S) -> T: ...
    async def athrow(self, typ: type, val: Any = None, tb: Any = None) -> T: ...

# Context manager protocols
class AsyncContextManager(Protocol[T]):
    async def __aenter__(self) -> T: ...
    async def __aexit__(self, exc_type, exc_val, exc_tb) -> Optional[bool]: ...

# Type guard protocol
class TypeGuard(Protocol):
    def __call__(self, obj: Any) -> bool: ...