Python: Shelling out safely from async functions

Calling an external command from an async function is very easy. Doing it without freezing the main event loop however takes a little additional care. This document aims to detail the basics of the process, rather than act as an exhaustive resource.

This guide is mainly written with web services like running python as a service via uvicorn from Google Cloud Run in mind.

The central rule

Do not call a blocking function inside async def.

Because subprocess.run() is completely synchronous and blocking, it halts the main event loop while it waits for the OS command to finish.

import subprocess

async def handle_request() -> str:
    result = subprocess.run(["bash", "script.sh"], capture_output=True, check=True, text=True, timeout=30,) 
    return result.stdout

This completely freezes the asyncio event loop and it cannot run unrelated coroutines. Uvicorn runs on a single thread by default. So, while subprocess.run() is executing, the entire uvicorn process goes deaf and blind. It cannot accept any other incoming HTTP requests. Other users trying to send requests will experience severe lag or timeouts.

So how do we avoid this?

Option 1: adapt the existing sync function with `asyncio.to_thread`

This can be a small and safe change, if the synchronous function already works well and has good error handling and doesn’t have a tendency to hang or take an exorbitant amount of time to execute.

import asyncio
import subprocess

def do_some_magic(source_path: str) -> str:
    result = subprocess.run(["bash", "script.sh"], capture_output=True, check=True, text=True, timeout=30,) 
    return result.stdout

async def do_some_magic_async(source_path: str) -> str:
    return await asyncio.to_thread(do_some_magic, source_path)

asyncio.to_thread() runs the sync function in the event loop’s default thread pool and returns an awaitable. The worker thread may block, but the event loop itself remains available to run other tasks, meaning that in this case, other HTTP requests can still come in and start getting processed.

This works well if you:

-> pass simple plain data in: strings or numbers preferrably. Indexed arrays (lists) and associative arrays (dictionaries) are also fine.
-> return simple plain data out: strings work the best
-> keep timeout and subprocess error handling inside the sync function
-> try not to chain a lot of long running functions within this subprocess.. short call outs to simple *nix tools like awk, sed, cat, grep, etc are usually fine

Timeout

timeout=30 raises a subprocess.TimeoutExpired which propagates through the await asyncio.to_thread() normally.

So you can put this stuff inside a try block:

async def do_some_magic_async(source_path: str) -> str:
    try:
        return await asyncio.to_thread(do_some_magic, source_path)
    except subprocess.TimeoutExpired as exc:
        #handle this somehow

Request Interruption

Basically, if the HTTP request is somehow interrupted, like it encounters a 500 error in the meantime, it doesn’t mean that the spawned bash process will die with it. So this is also a good reason for having a reasonable timeout set for these sync -> async situations.

Option 2: use the subprocess call natively available from asyncio

This is more complicated, but gives the event loop diret control of the child process that gets spawned:

import asyncio
import os

async def run_bash_script():
    # Path to your executable script
    script_path = "./do_magic.sh"
    
    script_args = ["magic_variable", "/var/www/html"]

    if not os.path.exists(script_path):
        print(f"Error: {script_path} not found.")
        return

    try:
        process = await asyncio.create_subprocess_exec(
            script_path,
            *script_args,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE
        )

        stdout, stderr = await process.communicate()

        print(f"Exit Code: {process.returncode}")
        if stdout:
            print(f"Output:\n{stdout.decode()}")
        if stderr:
            print(f"Errors:\n{stderr.decode()}")

    #Note: in google cloud run, most of the system will be set to read only, so do the chmod in the docker file
    except PermissionError: 
        print(f"Permission denied. Did you run 'chmod +x {script_path}'?")
    except Exception as e:
        print(f"An error occurred: {e}")

if __name__ == "__main__":
    asyncio.run(run_bash_script())

Key takeaways:

-> Stream Redirection: Setting stdout=asyncio.subprocess.PIPE redirects the child process’s output to an asynchronous stream reader accessible on the process object.
-> Avoid Deadlocks: Always use await process.communicate() rather than await process.wait() if you are using pipes. wait() can deadlock if the child process fills up the OS buffer with too much output before finishing. This can be particularly important if dealing with bigger chunks of inputs, like user uploaded csv, tsv, xml, etc files
-> Arguments Array: Unlike create_subprocess_shell, which takes a single command string, create_subprocess_exec expects individual command-line arguments separated as distinct strings (e.g., ['ls', '-l', '/tmp']). This adds some mitigation against shell injection vulnerabilities.
-> env vars and current working directory are inherited from the parent python script, but the bash script does independently have access to OS level env variables

Conclusion

In an environment like google cloud run, “shelling out” can be a little more complicated than the initial idea suggests. Hope this writeup is helpful as a basic orientation.