¿Cómo uso subprocess.Popen para conectar múltiples procesos por tuberías?

https://stackoverflow.com/questions/295459

08-07-2019
|

Pregunta

¿Cómo ejecuto el siguiente comando de shell utilizando Python subproceso módulo?

echo "input data" | awk -f script.awk | sort > outfile.txt

Los datos de entrada provendrán de una cadena, por lo que no necesito echo . He llegado hasta aquí, ¿alguien puede explicar cómo hago que se canalice a través de sort también?

p_awk = subprocess.Popen(["awk","-f","script.awk"],
                          stdin=subprocess.PIPE,
                          stdout=file("outfile.txt", "w"))
p_awk.communicate( "input data" )

ACTUALIZACIÓN : Tenga en cuenta que aunque la respuesta aceptada a continuación en realidad no responde la pregunta como se le preguntó, creo que S.Lott tiene razón y es mejor evitar tener que resolver ese problema en primer lugar !

Solución

Serías un poco más feliz con lo siguiente.

import subprocess

awk_sort = subprocess.Popen( "awk -f script.awk | sort > outfile.txt",
    stdin=subprocess.PIPE, shell=True )
awk_sort.communicate( b"input data\n" )

Delegar parte del trabajo al shell. Deje que conecte dos procesos con una tubería.

Sería mucho más feliz reescribir 'script.awk' en Python, eliminando awk y la tubería.

Editar . Algunas de las razones para sugerir que awk no está ayudando.

[Hay demasiadas razones para responder mediante comentarios.]

Awk está agregando un paso sin valor significativo. No hay nada único sobre el procesamiento de awk que Python no maneje.
La canalización de awk a sort, para grandes conjuntos de datos, puede mejorar el tiempo de procesamiento transcurrido. Para conjuntos cortos de datos, no tiene un beneficio significativo. Una medición rápida de awk > file; ordenar archivo y awk | sort revelará las ayudas de concurrencia. Con sort, rara vez ayuda porque sort no es un filtro de un solo paso.
La simplicidad de " Python para ordenar " el procesamiento (en lugar de "Python to awk to sort") evita el tipo exacto de preguntas que se hacen aquí.
Python, aunque es más verbal que awk, también es explícito cuando awk tiene ciertas reglas implícitas que son opacas para los novatos y confusas para los no especialistas.
Awk (como el script de shell en sí) agrega otro lenguaje de programación. Si todo esto se puede hacer en un lenguaje (Python), la eliminación de la programación de shell y awk elimina dos lenguajes de programación, lo que permite a alguien concentrarse en las partes de la tarea que generan valor.

Conclusión: awk no puede agregar un valor significativo. En este caso, awk es un costo neto; agregaba suficiente complejidad que era necesario hacer esta pregunta. Eliminar awk será una ganancia neta.

Barra lateral Por qué es tan difícil construir una tubería ( a | b ).

Cuando el shell se enfrenta con a | b tiene que hacer lo siguiente.

Bifurca un proceso hijo del shell original. Esto eventualmente se convertirá en b.
Construye una tubería os. (no es un subproceso Python.PIPE) pero llame a os.pipe () que devuelve dos nuevos descriptores de archivo que están conectados a través de un buffer común. En este punto, el proceso tiene stdin, stdout, stderr de su padre, más un archivo que será "stdout de a". y "stdin de b".
Bifurca a un niño. El niño reemplaza su stdout con el nuevo stdout de a. Ejecute el proceso a .
El b niño cierra reemplaza su stdin con el nuevo b's stdin. Ejecute el proceso b .
El niño b espera a que se complete a.
El padre está esperando que se complete b.

Creo que lo anterior se puede usar de forma recursiva para generar a | b | c , pero debe paréntesis implícitamente las canalizaciones largas, tratándolas como si fueran a | (b | c) .

Dado que Python tiene os.pipe () , os.exec () y os.fork () , y puede reemplazar < code> sys.stdin y sys.stdout , hay una manera de hacer lo anterior en Python puro. De hecho, es posible que pueda resolver algunos atajos utilizando os.pipe () y subprocess.Popen .

Sin embargo, es más fácil delegar esa operación al shell.

Otros consejos

import subprocess

some_string = b'input_data'

sort_out = open('outfile.txt', 'wb', 0)
sort_in = subprocess.Popen('sort', stdin=subprocess.PIPE, stdout=sort_out).stdin
subprocess.Popen(['awk', '-f', 'script.awk'], stdout=sort_in, 
                 stdin=subprocess.PIPE).communicate(some_string)

Para emular una tubería de shell:

from subprocess import check_call

check_call('echo "input data" | a | b > outfile.txt', shell=True)

sin invocar el shell (consulte 17.1.4.2. Reemplazo del shell canalización ):

#!/usr/bin/env python
from subprocess import Popen, PIPE

a = Popen(["a"], stdin=PIPE, stdout=PIPE)
with a.stdin:
    with a.stdout, open("outfile.txt", "wb") as outfile:
        b = Popen(["b"], stdin=a.stdout, stdout=outfile)
    a.stdin.write(b"input data")
statuses = [a.wait(), b.wait()] # both a.stdin/stdout are closed already

plumbum proporciona algo de azúcar de sintaxis:

#!/usr/bin/env python
from plumbum.cmd import a, b # magic

(a << "input data" | b > "outfile.txt")()

El análogo de:

#!/bin/sh
echo "input data" | awk -f script.awk | sort > outfile.txt

es:

#!/usr/bin/env python
from plumbum.cmd import awk, sort

(awk["-f", "script.awk"] << "input data" | sort > "outfile.txt")()

http://www.python.org/doc/2.5.2/lib/node535.html cubrió esto bastante bien. ¿Hay alguna parte de esto que no entendiste?

Su programa sería bastante similar, pero el segundo Popen tendría stdout = en un archivo, y no necesitaría la salida de su .communicate () .

Inspirado por la respuesta de @ Cristian. Me encontré con el mismo problema, pero con un comando diferente. Así que estoy poniendo mi ejemplo probado, que creo que podría ser útil:

grep_proc = subprocess.Popen(["grep", "rabbitmq"],
                             stdin=subprocess.PIPE, 
                             stdout=subprocess.PIPE)
subprocess.Popen(["ps", "aux"], stdout=grep_proc.stdin)
out, err = grep_proc.communicate()

Esto está probado.

Lo que se ha hecho

Se declaró ejecución perezosa de grep con stdin de pipe. Este comando se ejecutará en la ejecución del comando ps cuando la tubería se llene con la salida estándar de ps .
Llamado al comando principal ps con stdout dirigido a la tubería utilizada por el comando grep .
Grep se comunicó para obtener stdout de la tubería.

Me gusta de esta manera porque es una concepción de tubería natural envuelta suavemente con interfaces subprocess .

La respuesta aceptada es esquivar el problema. Aquí hay un fragmento que encadena la salida de múltiples procesos: Tenga en cuenta que también imprime el comando de shell (algo) equivalente para que pueda ejecutarlo y asegurarse de que la salida sea correcta.

#!/usr/bin/env python3

from subprocess import Popen, PIPE

# cmd1 : dd if=/dev/zero bs=1m count=100
# cmd2 : gzip
# cmd3 : wc -c
cmd1 = ['dd', 'if=/dev/zero', 'bs=1M', 'count=100']
cmd2 = ['tee']
cmd3 = ['wc', '-c']
print(f"Shell style : {' '.join(cmd1)} | {' '.join(cmd2)} | {' '.join(cmd3)}")

p1 = Popen(cmd1, stdout=PIPE, stderr=PIPE) # stderr=PIPE optional, dd is chatty
p2 = Popen(cmd2, stdin=p1.stdout, stdout=PIPE)
p3 = Popen(cmd3, stdin=p2.stdout, stdout=PIPE)

print("Output from last process : " + (p3.communicate()[0]).decode())

# thoretically p1 and p2 may still be running, this ensures we are collecting their return codes
p1.wait()
p2.wait()
print("p1 return: ", p1.returncode)
print("p2 return: ", p2.returncode)
print("p3 return: ", p3.returncode)

EDITAR: pipes está disponible en Windows pero, crucialmente, no parece que realmente funcione en Windows. Vea los comentarios a continuación.

La biblioteca estándar de Python ahora incluye el módulo pipes para manejar esto:

https://docs.python.org/2/library/pipes.html , https://docs.python.org/3.4/library/pipes.html

No estoy seguro de cuánto tiempo ha estado presente este módulo, pero este enfoque parece ser mucho más simple que burlarse del subproceso .

Las respuestas anteriores perdieron un punto importante. Reemplazar tubería de shell es básicamente correcto, como se señaló por geocar. Es casi suficiente para ejecutar comunicar en el último elemento de la tubería.

El problema restante es pasar los datos de entrada a la tubería. Con múltiples subprocesos, un simple comunicar (input_data) en el último elemento no funciona, se cuelga para siempre. Debe crear una tubería y un elemento secundario manualmente de esta manera:

import os
import subprocess

input = """\
input data
more input
""" * 10

rd, wr = os.pipe()
if os.fork() != 0: # parent
    os.close(wr)
else:              # child
    os.close(rd)
    os.write(wr, input)
    os.close(wr)
    exit()

p_awk = subprocess.Popen(["awk", "{ print $2; }"],
                         stdin=rd,
                         stdout=subprocess.PIPE)
p_sort = subprocess.Popen(["sort"], 
                          stdin=p_awk.stdout,
                          stdout=subprocess.PIPE)
p_awk.stdout.close()
out, err = p_sort.communicate()
print (out.rstrip())

Ahora el hijo proporciona la entrada a través de la tubería y el padre llama a comunicar (), que funciona como se esperaba. Con este enfoque, puede crear tuberías largas arbitrarias sin recurrir a "delegar parte del trabajo al shell". Lamentablemente, la documentación del subproceso no menciona esto.

Hay formas de lograr el mismo efecto sin tuberías:

from tempfile import TemporaryFile
tf = TemporaryFile()
tf.write(input)
tf.seek(0, 0)

Ahora use stdin = tf para p_awk . Es cuestión de probar lo que prefieres.

Lo anterior aún no es 100% equivalente a las tuberías de bash porque el manejo de la señal es diferente. Puede ver esto si agrega otro elemento de tubería que trunca la salida de sort , p. head -n 10 . Con el código anterior, sort imprimirá un " Tubo roto " mensaje de error a stderr . No verá este mensaje cuando ejecute la misma tubería en el shell. (Sin embargo, esa es la única diferencia, el resultado en stdout es el mismo). La razón parece ser que el Popen de python establece SIG_IGN para SIGPIPE , mientras que el shell lo deja en SIG_DFL , y El manejo de la señal de sort es diferente en estos dos casos.

Para mí, el siguiente enfoque es el más limpio y fácil de leer

from subprocess import Popen, PIPE

def string_to_2_procs_to_file(input_s, first_cmd, second_cmd, output_filename):
    with open(output_filename, 'wb') as out_f:
        p2 = Popen(second_cmd, stdin=PIPE, stdout=out_f)
        p1 = Popen(first_cmd, stdout=p2.stdin, stdin=PIPE)
        p1.communicate(input=bytes(input_s))
        p1.wait()
        p2.stdin.close()
        p2.wait()

que se puede llamar así:

string_to_2_procs_to_file('input data', ['awk', '-f', 'script.awk'], ['sort'], 'output.txt')

Licenciado bajo: CC-BY-SA con atribución

No afiliado a StackOverflow