¿Cómo escribo una versión asíncrona de subprocess.check_output?

Answer

1. 

   davy.ai

   June 5, 2023 at 2:22 pm

   Para escribir un equivalente asincrónico de subprocess.check_output y capturar la salida completa de la ejecución de un proceso en el marco de trabajo asyncio, se puede usar la función asyncio.create_subprocess_exec() para crear y ejecutar el comando de forma asíncrona.

   async def check_output(cmd):
       process = await asyncio.create_subprocess_exec(*cmd,
                                                      stdout=asyncio.subprocess.PIPE,
                                                      stderr=asyncio.subprocess.PIPE)
       output, error = await process.communicate()
       if process.returncode == 0:
           return output.decode().strip()
       else:
           error_message = error.decode().strip() if error else ''
           raise subprocess.CalledProcessError(process.returncode, cmd, error_message)
   

   En el código anterior, creamos una corutina check_output que toma un comando (como una lista de cadenas) como argumento. Luego creamos un subproceso usando asyncio.create_subprocess_exec() y pasamos el comando a él. También especificamos stdout y stderr como asyncio.subprocess.PIPE para capturar la salida y el error del subprocess.

   Luego usamos await process.communicate() para ejecutar el subprocesso y capturar su salida y error. Si el subprocesso devuelve un código de salida diferente a cero, lanzamos un subprocess.CalledProcessError con el comando y el mensaje de error.

   De lo contrario, devolvemos la salida capturada por el subprocesso. Convertimos la salida de bytes a una cadena usando decode(), y eliminamos cualquier espacio en blanco inicial o final con strip().

Comments are closed.