How can one simulate the behavior of "subprocess.check_call" but with asyncio?
asyncio seems to provide only the method "asyncio.create_subprocess_shell", which is lower level.
It's actually quite simple to wrap create_subprocess_shell to act like check_call. From the specification in python docs:
Run command with arguments. Wait for command to complete. If the return code was zero then return, otherwise raise CalledProcessError. The CalledProcessError object will have the return code in the returncode attribute. If check_call() was unable to start the process it will propagate the exception that was raised.
So we can write this function
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)
so that in a case like this
async def main():
await check_call("ls non_existing_path")
asyncio.run(main())
will print the following
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.
which is quite similar to how python's check_call would behave.
If you love us? You can donate to us via Paypal or buy me a coffee so we can maintain and grow! Thank you!
Donate Us With