每个PEP-492我试图实现一个异步迭代器,这样我就可以做到
async for foo in bar:
...
这是一个简单的例子,类似于文档中的例子,对实例化和异步迭代进行了非常基本的测试:
import pytest
class TestImplementation:
def __aiter__(self):
return self
async def __anext__(self):
raise StopAsyncIteration
@pytest.mark.asyncio # note use of pytest-asyncio marker
async def test_async_for():
async for _ in TestImplementation():
pass
但是,当我执行我的测试套件时,我看到:
=================================== FAILURES ===================================
________________________________ test_async_for ________________________________
@pytest.mark.asyncio
async def test_async_for():
> async for _ in TestImplementation():
E TypeError: 'async for' received an invalid object from __aiter__: TestImplementation
...: TypeError
===================== 1 failed, ... passed in 2.89 seconds ======================
为什么我的TestImplementation
似乎无效?据我所知,它符合协议:
- 对象必须实现
__aiter__
方法...返回异步迭代器对象。- 异步迭代器对象必须实现一个
__anext__
方法...返回一个等待的。- 要停止迭代,
__anext__
必须引发StopAsyncIteration
异常。
最新发布的Python版本(3.5.1),py.test
(2.9.2)和pytest-asyncio
(0.4.1)都失败了。
如果你读a little further down the documentation它提到(强调我的):
PEP 492在CPython 3.5.0中被接受,其中
__aiter__
被定义为一种方法,期望返回一个等待解析为异步迭代器的方法。在3.5.2中(由于临时接受PEP 492),更新了
__aiter__
协议以直接返回异步迭代器。
因此,对于3.5.2之前的版本(2016/6/27发布),文档与如何编写工作异步迭代器略有不同。 3.5.0和3.5.1的固定版本如下所示:
class TestImplementation:
async def __aiter__(self):
# ^ note
return self
async def __anext__(self):
raise StopAsyncIteration
这是在关闭bug #27243时引入的,并且在data model documentation中稍微清楚一点,这也提出了一种编写向后兼容代码的方法。
异步迭代器已在Python 3.6中实现 - 请参阅PEP-525
然后你根本不需要你的TestImplementation来使用async for
。你可以使用yield
(例子来自PEP-525):
async def ticker(delay, to):
"""Yield numbers from 0 to `to` every `delay` seconds."""
for i in range(to):
yield i
await asyncio.sleep(delay)
然后你可以按照预期使用async for
:
async for i in ticker(1, 10):
print(f'Tick #{i}')