Реализация асинхронного итератора

Per 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).

Ответы

Ответ 1

Если вы читаете чуть дальше документацию, в нем упоминается, что (внимание мое):

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

Это было введено при закрытии ошибки # 27243 и немного яснее в документации модели данных, который также предлагает способ записи обратно совместимого кода.

Ответ 2

Асинхронные итераторы были реализованы в 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}')