La forma correcta de ejecutar comandos Shell desde Python

Estas son todas las opciones que tienes en Python para ejecutar otros procesos - lo malo, lo bueno, y lo más importante, la forma correcta de hacerlo.

Python es una opción popular para automatizar cualquier cosa y todo, eso incluye automatizar tareas de administración del sistema o tareas que requieren ejecutar otros programas o interactuar con el sistema operativo. Sin embargo, hay muchas maneras de lograr esto en Python, la mayoría de las cuales son discutiblemente malas.

Así que, en este artículo, veremos todas las opciones de Python para ejecutar otros procesos - las malas, las buenas, y lo más importante, la forma correcta de hacerlo.

Las Opciones

Python tiene demasiadas opciones incorporadas para interactuar con otros programas, algunas de ellas mejores o peores, y honestamente, no me gusta ninguna de ellas. Echemos un vistazo rápido a cada opción y veamos cuándo (si es que alguna vez) tiene sentido usar el módulo en particular.

Herramientas nativas

La regla general debería ser usar funciones nativas en lugar de llamar directamente a otros programas o comandos del SO. Así que, primero, veamos las opciones nativas de Python:

Pathlib: Si necesitas crear o borrar un archivo/directorio; comprobar si un archivo existe; cambiar permisos; etc, no hay absolutamente ninguna razón para ejecutar comandos del sistema. Simplemente usa pathlib, tiene todo lo que necesitas. Cuando empieces a usar pathlib, también te darás cuenta de que puedes olvidarte de otros módulos de Python, como glob, u os.path.
Tempfile: Del mismo modo, si necesitas un archivo temporal sólo tienes que utilizar el módulo tempfile, no te líes con /tmp manualmente.
Shutil - pathlib: Debería satisfacer la mayoría de tus necesidades relacionadas con ficheros en Python, pero si necesitas, por ejemplo, copiar, mover, chown, which o crear un archivo, entonces deberías recurrir a shutil.
Signal: En caso de que necesites usar manejadores de signals.
Syslog: Para una interfaz a Unix syslog.
Si ninguna de las opciones incorporadas arriba satisface sus necesidades, sólo entonces tiene sentido empezar a interactuar con el SO u otros programas directamente.

Módulo OS

Empezando por las peores opciones módulo os proporciona funciones de bajo nivel para interactuar con OS, muchas de las cuales han sido reemplazadas por funciones en otros módulos.

Si simplemente quisieras llamar a algún otro programa, podrías usar la función os.system, pero no deberías. Ni siquiera quiero darte un ejemplo, porque simplemente no deberías usarla.

Aunque os no debería ser tu primera opción, hay un par de funciones que podrías encontrar útiles:

import os

print(os.getenv('PATH'))
# /home/martin/.local/bin:/usr/local/sbin:/usr/local/bin:...
print(os.uname())
# posix.uname_result(sysname='Linux', nodename='...', release='...', version='...', machine='x86_64')
print(os.times())
# posix.times_result(user=0.01, system=0.0, children_user=0.0, children_system=0.0, elapsed=1740.63)
print(os.cpu_count())
# 16
print(os.getloadavg())
# (2.021484375, 2.35595703125, 2.04052734375)
old_umask = os.umask(0o022)
# Do stuff with files...
os.umask(old_umask)  # restore old umask

# Only if you need better random numbers than pseudo-random numbers from 'random' module:
from base64 import b64encode

random_bytes = os.urandom(64)
print(b64encode(random_bytes).decode('utf-8'))
# C2F3kHjdzxcP7461ETRj/YZredUf+NH...hxz9MXXHJNfo5nXVH7e5olqLwhahqFCe/mzLQ==

Aparte de la función mostrada arriba, también hay funciones para crear fd (descriptores de fichero), pipes, abrir PTY, chroot, chmod, mkdir, kill, stat, pero me gustaría desanimarte a usarlas ya que hay mejores opciones. Incluso hay una sección en la documentación que muestra cómo reemplazar os con el módulo subproceso, así que ni se te ocurra usar os.popen, os.spawn o os.system.

Lo mismo ocurre con el uso del módulo os para operaciones de archivo/ruta - por favor, no lo hagas. Aquí hay toda una sección sobre cómo usar pathlib en lugar de os.path y otras funciones relacionadas con rutas.

La mayoría de las funciones restantes del módulo os son una interfaz directa a la API del SO (o del lenguaje C), por ejemplo os.dup, os.splice, os.mkfifo, os.execv, os.fork, etc. Si necesitas usar todos esos, entonces no estoy seguro de que Python sea el lenguaje adecuado para la tarea.

Módulo de subproceso

Una segunda opción un poco mejor que tenemos en Python es el módulo de subproceso. Esto es lo que parece:

import subprocess

p = subprocess.run('ls -l', shell=True, check=True, capture_output=True, encoding='utf-8')

# 'p' is instance of 'CompletedProcess(args='ls -la', returncode=0)'
print(f'Command {p.args} exited with {p.returncode} code, output: \n{p.stdout}')
# Command ls -la exited with 0 code

# total 36
# drwxrwxr-x  2 martin martin  4096 apr 22 12:53 .
# drwxrwxr-x 42 martin martin 20480 apr 22 11:01 ..
# ...

Como se indica en la documentación:

El enfoque recomendado para invocar subprocesos es utilizar la función run() para todos los casos de uso que puede manejar.

En la mayoría de los casos, debería ser suficiente para que usted utilice subprocess.run, pasando kwargs para alterar su comportamiento, por ejemplo, shell=True le permite pasar el comando como una sola cadena, check=True hace que lance una excepción si el código de salida no es 0, y capture_output=True rellena el atributo stdout.

Mientras que subprocess.run() es la forma recomendada de invocar procesos. Hay otras opciones (innecesarias, obsoletas) en este módulo: call, check_call, check_output, getstatusoutput, getoutput. Generalmente, usted debería usar sólo run y Popen, como se muestra a continuación:

with subprocess.Popen(['ls', '-la'], stdout=subprocess.PIPE, encoding='utf-8') as process:
    # process.wait(timeout=5)  # Returns only code: 0
    outs, errs = process.communicate(timeout=5)
    print(f'Command {process.args} exited with {process.returncode} code, output: \n{outs}')

# Pipes
import shlex
ls = shlex.split('ls -la')
awk = shlex.split("awk '{print $9}'")
ls_process = subprocess.Popen(ls, stdout=subprocess.PIPE)
awk_process = subprocess.Popen(awk, stdin=ls_process.stdout, stdout=subprocess.PIPE, encoding='utf-8')

for line in awk_process.stdout:
    print(line.strip())
    # .
    # ..
    # examples.py
    # ...

El primer ejemplo anterior muestra el equivalente en Popen del subproceso.run mostrado anteriormente. Sin embargo, sólo debe utilizar Popen cuando necesite más flexibilidad de la que proporciona run, por ejemplo, en el segundo ejemplo, puede ver cómo puede canalizar la salida de un comando a otro, ejecutando ls -la | awk '{print $9}'. También puede ver que usamos shlex.split, que es una función conveniente que divide la cadena en una matriz de tokens que pueden pasarse a Popen o ejecutarse sin usar shell=True.

Cuando usas Popen, puedes usar adicionalmente terminate(), kill() y send_signal() para más interacciones con el proceso.

En los ejemplos anteriores, realmente no hicimos ningún manejo de errores, pero muchas cosas pueden salir mal cuando se ejecutan otros procesos. Para scripts simples, check=True es probablemente suficiente ya que causará que CalledProcessError sea levantado tan pronto como el subproceso se ejecute en un código de retorno distinto de cero, por lo que su programa fallará rápido y fuerte, lo cual es bueno. Si también estableces el argumento timeout, entonces también puedes obtener la excepción TimeoutExpired, pero generalmente, todas las excepciones en el módulo de subprocesos heredan de SubprocessError, así que si quieres atrapar excepciones, entonces puedes simplemente estar atento a SubprocessError.

La manera correcta

El Zen de Python establece que:

Debería haber una y preferiblemente sólo una forma obvia de hacerlo.

Pero, hasta ahora, hemos visto bastantes maneras, todas en los módulos incorporados de Python. Entonces, ¿cuál es la correcta? En mi opinión, ninguna.

Aunque me encanta la biblioteca estándar de Python, creo que una de las "pilas" que le faltan es un mejor módulo de subprocesos.

Si te encuentras orquestando muchos otros procesos en Python, entonces al menos deberías echarle un vistazo a la librería sh:

# https://pypi.org/project/sh/
# pip install sh
import sh

# Run any command in $PATH...
print(sh.ls('-la'))

ls_cmd = sh.Command('ls')
print(ls_cmd('-la'))  # Explicit
# total 36
# drwxrwxr-x  2 martin martin  4096 apr  8 14:18 .
# drwxrwxr-x 41 martin martin 20480 apr  7 15:23 ..
# -rw-rw-r--  1 martin martin    30 apr  8 14:18 examples.py

# If command is not in PATH:
custom_cmd = sh.Command('/path/to/my/cmd')
custom_cmd('some', 'args')

with sh.contrib.sudo:
    # Do stuff using 'sudo'...
    ...

Cuando invocamos sh.some_command, la biblioteca sh intenta buscar un comando shell incorporado o un binario en su $PATH con ese nombre. Si encuentra tal comando, simplemente lo ejecutará por ti. Si el comando no está en $PATH, entonces puede crear una instancia de Command y llamarlo de esa manera. En caso de que necesites usar sudo, puedes usar el gestor de contexto sudo del módulo contrib. Tan simple y directo, ¿verdad?

Para escribir la salida de un comando a un archivo, sólo necesitas proporcionar el argumento _out a la función:

sh.ip.address(_out='/tmp/ipaddr')
# Same as 'ip address > /tmp/ipaddr'

Lo anterior también muestra cómo invocar subcomandos: utilice puntos.

Y, por último, también puedes utilizar pipes (|) utilizando el argumento _in:

print(sh.awk('{print $9}', _in=sh.ls('-la')))
# Same as "ls -la | awk '{print $9}'"

print(sh.wc('-l', _in=sh.ls('.', '-1')))
# Same as "ls -1 | wc -l"

En cuanto a la gestión de errores, puede simplemente estar atento a las excepciones ErrorReturnCode o TimeoutException:

try:
    sh.cat('/tmp/doesnt/exist')
except sh.ErrorReturnCode as e:
    print(f'Command {e.full_cmd} exited with {e.exit_code}')
    # Command /usr/bin/cat /tmp/doesnt/exist exited with 1

curl = sh.curl('https://httpbin.org/delay/5', _bg=True)
try:
    curl.wait(timeout=3)
except sh.TimeoutException:
    print("Command timed out...")
    curl.kill()

Opcionalmente, si su proceso termina por una signal, recibirá SignalException, puedes comprobar signals específicas con, por ejemplo, SignalException_SIGKILL (o _SIGTERM, _SIGSTOP, etc.).

Esta librería también tiene soporte de registro integrado. Todo lo que tienes que hacer es activarlo. El siguiente código puede ayudar:

import logging

# Turn on default logging:
logging.basicConfig(level=logging.INFO)
sh.ls('-la')
# INFO:sh.command:<Command '/usr/bin/ls -la', pid 1631463>: process started

# Change log level:
logging.getLogger('sh').setLevel(logging.DEBUG)
sh.ls('-la')
# INFO:sh.command:<Command '/usr/bin/ls -la', pid 1631661>: process started
# DEBUG:sh.command:<Command '/usr/bin/ls -la'>: starting process
# DEBUG:sh.command.process:<Command '/usr/bin/ls -la'>.<Process 1631666 ['/usr/bin/ls', '-la']>: started process
# ...

Reflexiones finales

Quiero hacer hincapié de nuevo siempre debes preferir funciones nativas de Python en lugar de utilizar comandos del sistema. Además, siempre prefiere usar librerías cliente de terceros, como kubernetes-client o el SDK del proveedor de Cloud, en lugar de ejecutar comandos CLI directamente.

Eso, en mi opinión, se aplica incluso si vienes de un fondo SysAdmin y se sienten más cómodos con shell que Python. Y, por último, aunque Python es un lenguaje estupendo y mucho más robusto que shell, si necesitas encadenar demasiados programas/comandos, tal vez, sólo tal vez deberías escribir shell script en su lugar.

Fuente