async_generator 的块

Question

我可以按如下方式获取迭代器的块：

def get_chunks_it(l, n):
    """ Chunks an iterator `l` in size `n`
    Args:
        l (Iterator[Any]): an iterator
        n (int): size of
    Returns:
        Generator[Any]
    """
    iterator = iter(l)
    for first in iterator:
        yield itertools.chain([first], itertools.islice(iterator, n - 1))

现在假设我有一个异步生成器（python 3.6）：

async def generator():
    for i in range(0, 10):
        yield i
        await asyncio.sleep(1)

我怎样才能得到结果

async_generator

的块（假设大小为3，会产生[0,1,2]，[3,4,5]，[6,7,8]，[9]），以便我可以写：

async for chunk in get_chunk_it_async(generator(), 3):
    print(chunk)

Answer 1

您可以使用aiostream.stream.chunks：

from aiostream import stream

async def main():
    async for x in stream.chunks(generator(), 3):
        print(x)

输出：

[0, 1, 2]
[3, 4, 5]
[6, 7, 8]
[9]

请参阅项目页面和文档了解更多信息。

_{免责声明：我是项目维护者。}

Answer 2

由于 Python 3.6 中缺乏

aiter()

函数，这有点复杂（一旦从

__aiter__

返回可等待的值被正确弃用，它将在 3.7 中添加）。

itertools

对象还没有异步版本。

定义您自己的：

try:
    aiter
except NameError:
    # not yet a built-in, define our own shim for now
    from inspect import iscoroutinefunction as _isasync
    def aiter(ob, _isasync=_isasync):
        obtype = type(ob)  # magic methods are looked up on the type
        if not hasattr(obtype, '__aiter__'):
            raise TypeError(f'{obtype.__name__!r} object is not async iterable')
        async_iter = obtype.__aiter__(ob)
        if _isasync(async_iter):
            # PEP 492 allowed for __aiter__ to be a coroutine, but 525 reverses this again
            raise TypeError(f'{obtype.__name__!r} object is not async iterable')
        return async_iter
    del _isasync

接下来，您需要定义异步

islice

和

chain

对象：

class achain():
    """Chain over multiple async iterators"""
    def __init__(self, *async_iterables):
        self._source = iter(async_iterables)
        self._active = None
    def __aiter__(self):
        return self
    async def __anext__(self):
        if self._source is None:
            # we are all done, nothing more to produce
            raise StopAsyncIteration
        if self._active is None:
            # get next async iterable to loop over
            ait = next(self._source, None)
            if ait is None:
                # we are all done, nothing more to produce
                self._source = None
                raise StopAsyncIteration
            self._active = aiter(ait)
        try:
            return await type(ait).__anext__(ait)
        except StopAsyncIteration:
            # exhausted, recurse
            self._active = None
            return await self.__anext__()

class aslice():
    """Slice an async iterator"""
    def __init__(self, ait, start, stop=None, step=1):
        if stop is None:
            start, stop = 0, start
        self._ait = ait
        self._next, self._stop, self._step = start, stop, step
        self._cnt = 0
    def __aiter__(self):
        return self
    async def __anext__(self):
        ait, stop = self._ait, self._stop
        if ait is None:
            raise StopAsyncIteration
        anext = type(ait).__anext__
        while self._cnt < self._next:
            try:
                await anext(ait)
            except StopAsyncIteration:
                self._ait = None
                raise
            self._cnt += 1
        if stop is not None and self._cnt >= stop:
            self._ait = None
            raise StopAsyncIteration
        try:
            item = await anext(ait)
        except StopAsyncIteration:
            self._ait = None
            raise
        self._cnt += 1
        self._next += self._step
        return item

完成这些后，只需在正确的位置添加

async

即可：

async def get_chunks_it(l, n):
    """ Chunks an async iterator `l` in size `n`
    Args:
        l (Iterator[Any]): an iterator
        n (int): size of
    Returns:
        Generator[Any]
    """
    iterator = aiter(l)
    async for first in iterator:
        async def afirst():
            yield first
        yield achain(afirst, aslice(iterator, n - 1))

Answer 3

使用

python3.10

，无需任何额外的库：

from typing import AsyncIterator, TypeVar

T = TypeVar("T")

async def chunks(async_iterator: AsyncIterator[T], size: int) -> AsyncIterator[list[T]]:
    """Group async iterator results to chunks."""
    while True:
        results = []
        for _ in range(size):
            try:
                result = await anext(async_iterator)
            except StopAsyncIteration:
                if results:
                    yield results
                return
            else:
                results.append(result)

        yield results