Python: async check_call

Question

如何模拟"subprocess.check_call"但是使用asyncio？

asyncio似乎只提供了方法"；asyncio.create_subprocess_shell"；，这是较低的级别。

Answer 1

实际上，将create_subprocess_shell包装为类似check_call的行为非常简单。来自python文档中的规范：

使用参数运行命令。等待命令完成。如果返回代码为零，则返回，否则引发CalledProcessError。CalledProcessError对象的returncode属性中将包含返回代码。如果check_call((无法启动进程，它将传播引发的异常。

所以我们可以编写这个函数

import asyncio
from subprocess import CalledProcessError
async def check_call(cmd, **kwargs):
process = await asyncio.create_subprocess_shell(cmd, **kwargs)
return_code = await process.wait()
if return_code != 0:
raise CalledProcessError(return_code, cmd)

所以在这种的情况下

async def main():
await check_call("ls non_existing_path")
asyncio.run(main())

将打印以下

ls: cannot access 'non_existing_path': No such file or directory
Traceback (most recent call last):
File "73214383.py", line 13, in <module>
asyncio.run(main())
File "/usr/lib/python3.8/asyncio/runners.py", line 44, in run
return loop.run_until_complete(main)
File "/usr/lib/python3.8/asyncio/base_events.py", line 616, in run_until_complete
return future.result()
File "73214383.py", line 11, in main
await check_call("ls non_existing_path")
File "73214383.py", line 8, in check_call
raise CalledProcessError(return_code, cmd)
subprocess.CalledProcessError: Command 'ls non_existing_path' returned non-zero exit status 2.

这与python的CCD_ 3的行为非常相似。