是否可以从函数将执行返回到事件循环.任务一旦完成将返回功能并继续执行?
我正在尝试使用pytest-asyncio插件
例:
@pytest.mark.asyncio
async def test_async1(event_loop):
print('start 1')
res = event_loop.create_task(
send_async_request("http://test.com", limit=1000))) # here I need to return execution to event loop and continue only after getting response from send_async_request function
print('end1',res)
@pytest.mark.asyncio
async def test_async2(event_loop):
print('start 2')
res = event_loop.create_task(
send_async_request("http://test2", limit=1000))) # here I need to return execution to event loop and continue only after getting response from send_async_request function
print('end2', res)
send_async_request- aiohttp:
@asyncio.coroutine
def send_async_request(url, method='GET'):
with aiohttp.ClientSession() as session:
resp = yield from session.get(url, timeout=60)
if resp.status == 200:
return resp.status, response
else:
return resp.status, False
解决方法:
当您的测试用pytest.mark.asyncio标记时,它们将成为协程,因此您可以使用await语法:
@pytest.mark.asyncio
async def test_sleep(event_loop):
result = await asyncio.sleep(1, result=3, loop=event_loop)
assert result == 3
编辑:另一个示例,具有多个睡眠操作:
@pytest.mark.asyncio
async def test_multiple_sleep(event_loop):
tasks = [event_loop.create_task(asyncio.sleep(1, result=x))
for x in range(10)]
results = await asyncio.gather(*tasks)
assert results == list(range(10))
原文地址:https://codeday.me/bug/20191025/1931479.html
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。