python - 是python3.6新改 'async for'不兼容enumerate

Question

我正在尝试将我的应用程序从 python2.7 移动到 python3.6，用于 asyncio 和相关库，但我发现某些功能无法正常工作。我使用电机从 mongodb 异步查询，例如:

async def do_query():
    song_already_processed = set()
    song_table = db.song_table
    async for index, item in enumerate(song_table.find({'lang': 'English'},
                              {'id': 1, '_id': 0, 'title': 1, 'artist.name': 1})):
        if index > 100:
            break
        if item['id'] in song_already_processed:
            continue
        song_already_processed.add(item['id'])

    print(len(song_already_processed))

但它引发了错误:

TypeError: 'AsyncIOMotorCursor' object is not iterable

我认为异步迭代器协议(protocol)不同于普通的迭代器协议(protocol)，因此枚举效果不佳。是否有我可以使用的异步枚举？

顺便说一句，我知道很多方法来获取 100 个文档并停止迭代，我只是想知道如何正确使用“async for”

Answer 1

asyncstdlib library (免责声明:我维护这个包)提供标准库助手的 async 变体。具体来说，asyncstdlib.enumerate像 enumerate 一样工作，但接受并生成一个异步可迭代对象。

import asyncstdlib as a

async for index, item in a.enumerate(song_table.find(...)):
    if index > 100:
        break
    ...

请注意，从异步迭代中中断通常不是一个好主意——迭代器可能不会在迭代结束时清理(参见 PEP 533 for details)。

由于您使用 index 只是为了获取前 100 个项目，您还可以使用 asyncstdlib.islice直接安全地限制迭代:

import asyncstdlib as a

async for item in a.islice(song_table.find(...), 100):
    ...

---

Python3.6 添加asynchronous generators ，它允许轻松实现异步枚举:

async def aenumerate(asequence, start=0):
    """Asynchronously enumerate an async iterator from a given start value"""
    n = start
    async for elem in asequence:
        yield n, elem
        n += 1

对于旧版本的 Python，必须手动具体化异步生成器:

class AsyncEnumerate:
    """Asynchronously enumerate an async iterator from a given start value"""
    def __init__(self, asequence, start=0):
        self._asequence = asequence
        self._value = start

    async def __anext__(self):
        elem = await self._asequence.__anext__()
        value, self._value = self._value, self._value + 1
        return value, elem

    def __aiter__(self):
        return self