python3.6新变化'async for'与枚举不兼容吗

Posted 2023-03-30

【中文标题】python3.6新变化\'async for\'与枚举不兼容吗

【英文标题】：Is python3.6 new change 'async for' not compatible with enumeratepython3.6新变化'async for'与枚举不兼容吗

【发布时间】：2018-01-03 03:43:33

【问题描述】：

我正在尝试将我的应用程序从 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

我认为异步迭代器协议与普通的迭代器协议不同，所以枚举不能很好地工作。有我可以使用的异步枚举吗？

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

【问题讨论】：

song_table.find 只是返回一个游标对象，为什么你认为它可以被迭代？

@Sraw TBF，光标似乎是“可迭代的”。当然看起来和..

异步迭代器和迭代器是两个不同的东西，围绕一个功能构建的工具可能不适用于另一个。据我所知，aenumerate 还没有正式的实现，但您可以查看this 以了解其结构

【参考方案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
    ...

请注意，将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

【参考方案2】：

如果你不介意有外部依赖，可以使用aiostream.stream.enumerate:

async for i, item in aiostream.stream.enumerate(cursor):
    ...

在demonstration 和documentation 中查看更多示例。

【讨论】：

AttributeError: module 'aiostream' has no attribute 'enumerate'

aiostream.enumerate => aiostream.stream.enumerate FTFY

【参考方案3】：

我有一个类似的问题，涉及将异步生成器用作使用枚举和异步 for 的可迭代对象。我意识到在我的情况下最简单的答案是通过修改 async generator 来返回索引，而不是 async for。见下文：

async def whitespace(data: List) -> tuple:
    """function to remove any whitespace in returned values"""
    for i, v in enumerate(data):
        yield i, x.strip()

async for i, x in whitespace(a_list_of_stuff):
    a_list_of_stuff[i] = x