异步生成器说它没有实现__anext__，当它这样做时

Question

第一次使用Aync生成器。我使用的是Python 3.9。

这是我的实现：

import asyncio
class SubEventStream():
def __init__(self) -> None:
self.queue = asyncio.Queue()
return
async def __aiter__(self):
return self
async def __anext__(self):
return await self.pop()
async def append(self, request):
return await self.queue.put(request)
async def pop(self):
r = await self.queue.get()
self.queue.task_done()
return r
def create_append_tasks(ls, q):
return [
asyncio.create_task(q.append(i))
for i in ls
]
async def append_tasks(q):
tasks = create_append_tasks(('a', 'b', 'c', 'd', 'e'), q)
return await asyncio.gather(*tasks)

async def run():
q = SubEventStream()
await append_tasks(q)
async for v in q:
print(v)
asyncio.run(run())

令人困惑的是，这是我一直得到的结果：

/tmp/tmp.ie3Dj7Q9hn/test.py:37: RuntimeWarning: coroutine 'SubEventStream.__aiter__' was never awaited
async for v in q:
RuntimeWarning: Enable tracemalloc to get the object allocation traceback
Traceback (most recent call last):
File "test.py", line 40, in <module>
asyncio.run(run())
File "/usr/lib/python3.9/asyncio/runners.py", line 44, in run
return loop.run_until_complete(main)
File "/usr/lib/python3.9/asyncio/base_events.py", line 642, in run_until_complete
return future.result()
File "test.py", line 37, in run
async for v in q:
TypeError: 'async for' received an object from __aiter__ that does not implement __anext__: coroutine

很明显，我确实实现了__anext__。什么事耽搁了？

Answer 1

__aiter__必须是常规方法，而不是async方法：

def __aiter__(self):
return self

请参阅：https://docs.python.org/3.9/reference/datamodel.html#asynchronous-迭代器

Answer 2

为了更好地理解这里的Async Generator:的简单示例

import asyncio
class AsyncCounter:
def __init__(self, n):
self.n = n
self.current = 0
async def __aiter__(self):
return self
async def __anext__(self):
if self.current < self.n:
await asyncio.sleep(1)  # Simulate some async work
result = self.current
self.current += 1
return result
else:
raise StopAsyncIteration
async def main():
async for i in AsyncCounter(5):
print(i)
asyncio.run(main())

在上面的代码中，async for语句对__aiter__返回的迭代器对象调用__anext__方法，该方法等待一秒钟并返回self.current的当前值。循环将继续，直到引发StopAsyncIteration异常，这表示迭代结束。

什么是async for？

async for语句的工作原理与常规for语句类似，但它用于通过重复调用__anext__来迭代异步迭代器，而不是常规迭代器。