Per PEP-492 I am trying to implement an asynchronous iterator, such that I can do e.g.
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
- An object must implement an
method ... returning an asynchronous iterator object.__aiter__
- An asynchronous iterator object must implement an
method ... returning an awaitable.__anext__
- To stop iteration
must raise a__anext__
exception.StopAsyncIteration
py.test
pytest-asyncio
If you read a little further down the documentation it mentions that (emphasis mine):
PEP 492 was accepted in CPython 3.5.0 with
__aiter__
defined as a method, that was expected to return an awaitable resolving to an asynchronous iterator.In 3.5.2 (as PEP 492 was accepted on a provisional basis) the
__aiter__
protocol was updated to return asynchronous iterators directly.
Therefore for versions prior to 3.5.2 (released 2016/6/27) the documentation is slightly out of step with how to write a working asynchronous iterator. The fixed version for 3.5.0 and 3.5.1 looks like:
class TestImplementation:
async def __aiter__(self):
# ^ note
return self
async def __anext__(self):
raise StopAsyncIteration
This was introduced on closing bug #27243 and is a little clearer in the data model documentation, which also suggests a way of writing backwards compatible code.