link several Popen commands with pipes

Question

I know how to run a command using cmd = subprocess.Popen and then subprocess.communicate. Most of the time I use a string tokenized with shlex.split as 'argv' argument for Popen. Example with "ls -l":

import subprocess
import shlex
print subprocess.Popen(shlex.split(r'ls -l'), stdin = subprocess.PIPE, stdout = subprocess.PIPE, stderr = subprocess.PIPE).communicate()[0]

However, pipes seem not to work... For instance, the following example returns noting:

import subprocess
import shlex
print subprocess.Popen(shlex.split(r'ls -l | sed "s/a/b/g"'), stdin = subprocess.PIPE, stdout = subprocess.PIPE, stderr = subprocess.PIPE).communicate()[0]

Can you tell me what I am doing wrong please?

Thx

Lysozyme

Related Question: #295959 – Gelid

Answer 1

I think you want to instantiate two separate Popen objects here, one for 'ls' and the other for 'sed'. You'll want to pass the first Popen object's stdout attribute as the stdin argument to the 2nd Popen object.

Example:

p1 = subprocess.Popen('ls ...', stdout=subprocess.PIPE)
p2 = subprocess.Popen('sed ...', stdin=p1.stdout, stdout=subprocess.PIPE)
print p2.communicate()

You can keep chaining this way if you have more commands:

p3 = subprocess.Popen('prog', stdin=p2.stdout, ...)

See the subprocess documentation for more info on how to work with subprocesses.

Rust

Would you please show a full, working example of how to chain three processes together. For example, I am not sure how many times I need to call communicate and when to close stdin and stdout (docs mention closing stdout and the answer below shows closing stdin. Thanks. – Gelid

@Gelid you should only need to call communicate once on the last process, since you've spawned all of them and chained them together. So, if you have processes p1, p2, ...., pn, just call pn.communicate() at the end. – Erme

Answer 2

I've made a little function to help with the piping, hope it helps. It will chain Popens as needed.

from subprocess import Popen, PIPE
import shlex

def run(cmd):
  """Runs the given command locally and returns the output, err and exit_code."""
  if "|" in cmd:    
    cmd_parts = cmd.split('|')
  else:
    cmd_parts = []
    cmd_parts.append(cmd)
  i = 0
  p = {}
  for cmd_part in cmd_parts:
    cmd_part = cmd_part.strip()
    if i == 0:
      p[i]=Popen(shlex.split(cmd_part),stdin=None, stdout=PIPE, stderr=PIPE)
    else:
      p[i]=Popen(shlex.split(cmd_part),stdin=p[i-1].stdout, stdout=PIPE, stderr=PIPE)
    i = i +1
  (output, err) = p[i-1].communicate()
  exit_code = p[0].wait()

  return str(output), str(err), exit_code

output, err, exit_code = run("ls -lha /var/log | grep syslog | grep gz")

if exit_code != 0:
  print "Output:"
  print output
  print "Error:"
  print err
  # Handle error here
else:
  # Be happy :D
  print output

Silvern

Answer 3

shlex only splits up spaces according to the shell rules, but does not deal with pipes.

It should, however, work this way:

import subprocess
import shlex

sp_ls = subprocess.Popen(shlex.split(r'ls -l'), stdin = subprocess.PIPE, stdout = subprocess.PIPE, stderr = subprocess.PIPE)
sp_sed = subprocess.Popen(shlex.split(r'sed "s/a/b/g"'), stdin = sp_ls.stdout, stdout = subprocess.PIPE, stderr = subprocess.PIPE)
sp_ls.stdin.close() # makes it similiar to /dev/null
output = sp_ls.communicate()[0] # which makes you ignore any errors.
print output

according to help(subprocess)'s

Replacing shell pipe line
-------------------------
output=`dmesg | grep hda`
==>
p1 = Popen(["dmesg"], stdout=PIPE)
p2 = Popen(["grep", "hda"], stdin=p1.stdout, stdout=PIPE)
output = p2.communicate()[0]

HTH

Broadwater

The example uses the following line: p1.stdout.close() # Allow p1 to receive a SIGPIPE if p2 exits.. Should I utilize it as well? – Gelid

@Gelid Yes. If you don't, p1 will try to write and doesn't get notified about no one listening any longer. As your program still has this file open, but doesn't read, you get a deadlock. – Broadwater

Thanks for your answer. I found out that if I forget to close the stdin, the whole cascaded process will hang. – Absquatulate

@Absquatulate Of course. That's because the first process expects that more data come in, thus doesn't close its stdout as well, and that happens across the whole pipe. Just the same if you do cat | sort | head: if you don't press ^D, cat waits for more input, sort cannot start and head also not. – Broadwater

Answer 4

"""
Why don't you use shell

"""

def output_shell(line):

    try:
        shell_command = Popen(line, stdout=PIPE, stderr=PIPE, shell=True)
    except OSError:
        return None
    except ValueError:
        return None

    (output, err) = shell_command.communicate()
    shell_command.wait()
    if shell_command.returncode != 0:
        print "Shell command failed to execute"
        return None
    return str(output)

Vullo

Answer 5

Thank @hernvnc, @glglgl, and @Jacques Gaudin for the answers. I fixed the code from @hernvnc. His version will cause hanging in some scenarios.

import shlex
from subprocess import PIPE
from subprocess import Popen
def run(cmd, input=None):
    """Runs the given command locally and returns the output, err and exit_code."""
    if "|" in cmd:        
        cmd_parts = cmd.split('|')
    else:
        cmd_parts = []
        cmd_parts.append(cmd)
    i = 0
    p = {}
    for cmd_part in cmd_parts:
        cmd_part = cmd_part.strip()
        if i == 0:
            if input:
                p[i]=Popen(shlex.split(cmd_part),stdin=PIPE, stdout=PIPE, stderr=PIPE)
            else:
                p[i]=Popen(shlex.split(cmd_part),stdin=None, stdout=PIPE, stderr=PIPE)
        else:
            p[i]=Popen(shlex.split(cmd_part),stdin=p[i-1].stdout, stdout=PIPE, stderr=PIPE)
        i = i +1
    # close the stdin explicitly, otherwise, the following case will hang.
    if input:
        p[0].stdin.write(input)
        p[0].stdin.close()
    (output, err) = p[i-1].communicate()
    exit_code = p[0].wait()
    return str(output), str(err), exit_code

# test case below
inp = b'[  CMServer State   ]\n\nnode        node_ip         instance state\n--------------------------------------------\n1  linux172 10.90.56.172    1        Primary\n2  linux173 10.90.56.173    2        Standby\n3  linux174 10.90.56.174    3        Standby\n\n[    ETCD State     ]\n\nnode        node_ip         instance state\n--------------------------------------------------\n1  linux172 10.90.56.172    7001     StateFollower\n2  linux173 10.90.56.173    7002     StateLeader\n3  linux174 10.90.56.174    7003     StateFollower\n\n[   Cluster State   ]\n\ncluster_state   : Normal\nredistributing  : No\nbalanced        : No\ncurrent_az      : AZ_ALL\n\n[  Datanode State   ]\n\nnode        node_ip         instance state            | node        node_ip         instance state            | node        node_ip         instance state\n------------------------------------------------------------------------------------------------------------------------------------------------------------------------\n1  linux172 10.90.56.172    6001     P Standby Normal | 2  linux173 10.90.56.173    6002     S Primary Normal | 3  linux174 10.90.56.174    6003     S Standby Normal'
cmd = "grep -E 'Primary' | tail -1 | awk '{print $3}'"

run(cmd, input=inp)

Absquatulate