如何获取异步生成器的第一项？答案

【问题标题】：How to get the first item of an async generator?如何获取异步生成器的第一项？
【发布时间】：2021-12-10 23:49:07
【问题描述】：

我有以下功能。

async def f(url):
    async with httpx.AsyncClient() as client:
        async with client.stream('GET', url) as resp:
            async for c in resp.aiter_bytes():
                yield c

我正在编写一个测试代码来测试函数是否产生。

@pytest.mark.asyncio
async def test():
    c = None
    async for c in f(URL):
        c = chunk
        break  # there will be no error if removing this line. However, I don't need to loop all the items
    assert c

但是，代码运行但出现以下警告/错误。什么错误？

lib\site-packages\httpx_client.py:2012：用户警告：未关闭。详情请见https://www.python-httpx.org/async/#opening-and-closing-clients。

“未关闭”警告是break引起的吗？

Exception ignored in: <function _ProactorBasePipeTransport.__del__ at 0x000001AC1D884DC0>
Traceback (most recent call last):
  File "C:\anaconda3\envs\s\lib\asyncio\proactor_events.py", line 116, in __del__
    self.close()
  File "C:\anaconda3\envs\s\lib\asyncio\proactor_events.py", line 108, in close
    self._loop.call_soon(self._call_connection_lost, None)
  File "C:\anaconda3\envs\s\lib\asyncio\base_events.py", line 719, in call_soon
    self._check_closed()
  File "C:\anaconda3\envs\s\lib\asyncio\base_events.py", line 508, in _check_closed
    raise RuntimeError('Event loop is closed')
RuntimeError: Event loop is closed

如果删除break 语句，错误就会消失。

【问题讨论】：

不，这个问题没有回答我的问题。
试试async with f(URL) as resp: for c in resp: ...
在async with f(URL) as resp上会得到AttributeError: __aexit__的错误。

标签： python python-asyncio

【解决方案1】：

在Async generator is not an iterator?的回答中，下面的代码将摆脱错误。但是，它需要调用一个-private-“特殊”函数__anext()__()。

@pytest.mark.asyncio
async def test():
    it = f(URL):
    c = it.__anext__()
    assert c

【讨论】：

我不想吹毛求疵，但__anext__ 是一种特殊方法，而不是私有方法。使用它没有任何问题。从 Python 3.10 开始，您可以改用 anext(it)（参见 here：“这调用了 async_iterator 的 __anext__() 方法”）。