测试从常规函数调用 python 协程(async def)

Posted

技术标签:

【中文标题】测试从常规函数调用 python 协程(async def)【英文标题】:Test calling a python coroutine (async def) from a regular function 【发布时间】:2017-05-06 22:45:47 【问题描述】:

假设我有一些异步协程,它获取一些数据并返回它。像这样:

async def fetch_data(*args):
  result = await some_io()
  return result

基本上这个协程是从协程链中调用的,初始协程是通过创建任务来运行的。 但是,如果出于测试目的,我只想在运行某个文件时以这种方式运行一个协程呢:

if __name__ == '__main__':
  result = await fetch_data(*args)
  print(result)

显然我不能这样做,因为我试图从非协程函数运行和等待协程。 所以问题是:是否有一些正确的方法可以从协程中获取数据而不调用它的函数? 我可以为result 制作一些Future 对象并等待它,但也许还有其他更简单、更清晰的方法?

【问题讨论】:

【参考方案1】:

你需要创建一个事件循环来运行你的协程:

import asyncio

async def async_func():
    return "hello"

loop = asyncio.get_event_loop()
result = loop.run_until_complete(async_func())
loop.close()

print(result)

或者作为一个函数:

def run_coroutine(f, *args, **kwargs):
    loop = asyncio.get_event_loop()
    result = loop.run_until_complete(f(*args, **kwargs))
    loop.close()
    return result

这样使用:

print(run_coroutine(async_func))

或者:

assert "expected" == run_coroutine(fetch_data, "param1", param2="foo")

【讨论】:

以上是关于测试从常规函数调用 python 协程(async def)的主要内容,如果未能解决你的问题,请参考以下文章

理解Python协程:从yield/send到yield from再到async/await

理解Python协程:从yield/send到yield from再到async/await

在 django 中使用 sync_to_async 时如何运行协程函数?

python爬虫--多任务异步协程, 快点,在快点......

python协程(4):asyncio

Python 协程详解,都在这里了