subprocess en Python: run, Popen, communicate, pipes y ejecución de comandos del sistema

El módulo subprocess de Python es la forma recomendada de lanzar procesos del sistema operativo desde tu código. Sustituye a las antiguas funciones os.system() y os.popen() con una API más segura, más flexible y con mejor manejo de errores. Desde Python 3.5, subprocess.run() cubre la mayoría de los casos; Popen queda para cuando necesitas control más fino sobre el proceso.

subprocess.run(): el caso más habitual

import subprocess

# Forma básica: ejecutar y esperar
resultado = subprocess.run(['ls', '-la', '/tmp'])
print(resultado.returncode)   # 0 si fue bien

# Capturar la salida
resultado = subprocess.run(
    ['python3', '--version'],
    capture_output=True,
    text=True          # decodifica stdout/stderr como str
)
print(resultado.stdout)   # Python 3.13.0n
print(resultado.stderr)   # (vacío)

check=True: lanzar excepción si falla

import subprocess

try:
    resultado = subprocess.run(
        ['false'],         # comando que siempre devuelve código 1
        check=True,        # lanza CalledProcessError si returncode != 0
        capture_output=True,
        text=True
    )
except subprocess.CalledProcessError as e:
    print(f"Falló con código {e.returncode}")
    print(f"stderr: {e.stderr}")

timeout: matar el proceso si tarda demasiado

import subprocess

try:
    resultado = subprocess.run(
        ['sleep', '10'],
        timeout=2.0    # TimeoutExpired si supera los 2 segundos
    )
except subprocess.TimeoutExpired:
    print("El proceso tardó demasiado y fue terminado")

input: enviar datos al stdin del proceso

import subprocess

resultado = subprocess.run(
    ['grep', 'error'],
    input="línea normalnuna línea con errornotra línean",
    capture_output=True,
    text=True
)
print(resultado.stdout)   # una línea con error

Popen: control en tiempo real del proceso

Usa Popen cuando necesitas leer la salida mientras el proceso aún está corriendo, escribir varias veces al stdin o implementar un timeout personalizado:

import subprocess

with subprocess.Popen(
    ['ping', '-c', '4', '8.8.8.8'],
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    text=True
) as proceso:
    # Leer línea a línea en tiempo real
    for linea in proceso.stdout:
        print(f">> {linea}", end='')

    proceso.wait()
    print(f"Código de salida: {proceso.returncode}")

communicate(): enviar entrada y recoger toda la salida

import subprocess

# communicate() es la forma segura de stdin+stdout+stderr con Popen
with subprocess.Popen(
    ['sort', '-r'],
    stdin=subprocess.PIPE,
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    text=True
) as proceso:
    stdout, stderr = proceso.communicate(
        input="banananmanzanannaranjan",
        timeout=5
    )

print(stdout)   # naranjanmanzananbananan

poll(): comprobar si el proceso ha terminado

import subprocess, time

proceso = subprocess.Popen(
    ['sleep', '3'],
    stdout=subprocess.PIPE
)

while proceso.poll() is None:
    print("El proceso sigue corriendo...")
    time.sleep(0.5)

print(f"Proceso terminado con código {proceso.returncode}")

shlex.split(): dividir comandos de cadena de texto

import shlex, subprocess

# Nunca construyas el comando como string y uses shell=True con datos del usuario
comando_str = "grep -rn 'def ' /var/www --include='*.py'"
args = shlex.split(comando_str)
print(args)
# ['grep', '-rn', 'def ', '/var/www', "--include=*.py"]

resultado = subprocess.run(args, capture_output=True, text=True)
print(resultado.stdout[:200])

El peligro de shell=True

Con shell=True el comando se pasa a /bin/sh -c, lo que permite inyección de comandos si incluyes datos controlados por el usuario:

import subprocess

# PELIGROSO: si nombre_fichero viene del usuario podría contener "; rm -rf /"
nombre_fichero = input("Nombre del fichero: ")

# MAL — nunca hagas esto con datos externos
# subprocess.run(f"cat {nombre_fichero}", shell=True)

# BIEN — lista de argumentos, sin shell=True
subprocess.run(['cat', nombre_fichero], check=True)

Encadenar procesos con pipes

import subprocess

# Equivalente a: ps aux | grep python | wc -l
p1 = subprocess.Popen(['ps', 'aux'], stdout=subprocess.PIPE)
p2 = subprocess.Popen(['grep', 'python'], stdin=p1.stdout, stdout=subprocess.PIPE)
p1.stdout.close()   # permite que p1 reciba SIGPIPE si p2 termina antes
p3 = subprocess.Popen(['wc', '-l'], stdin=p2.stdout, stdout=subprocess.PIPE, text=True)
p2.stdout.close()

salida, _ = p3.communicate()
print(f"Procesos Python en ejecución: {salida.strip()}")

Variables de entorno para el subproceso

import subprocess, os

env_personalizado = {**os.environ, 'MI_VAR': 'valor_personalizado', 'DEBUG': '1'}

resultado = subprocess.run(
    ['printenv', 'MI_VAR'],
    env=env_personalizado,
    capture_output=True,
    text=True
)
print(resultado.stdout)   # valor_personalizado

La biblioteca subprocess cubre desde los casos más simples hasta pipelines complejos. La regla general: usa subprocess.run() con capture_output=True, check=True y text=True para el 80 % de los casos; recurre a Popen solo cuando necesitas interactuar con el proceso en tiempo real.

COMPARTE ESTE ARTÍCULO

COMPARTIR EN FACEBOOK
COMPARTIR EN TWITTER
COMPARTIR EN LINKEDIN
COMPARTIR EN WHATSAPP