https://golang.hateblo.jp/entry/python-subprocess

python の subprocess.run とコマンドのパイプ、subprocess.Popen について。

subprocess.run

subprocess.run は内部的には subprocess.Popen が使われていて Interrupt シグナルでプロセスを終了させたり、プロセスが終わるまで待ったりする処理を書かなくてもいいようにしてくれる。

コマンドを実行してそのまま出力する

subprocess.run(['ls', '-l'])

実行結果を出力せずに何かに使う

p = subprocess.run(['ls', '-l'], capture_output=True)
text = p.stdout.decode('utf-8')

# 予めbyte列からプレーンテキストにするなら universal_newlines(textと同じ) を true にする
subprocess.run(['ls', '-l'],
               encoding='utf-8',
               universal_newlines=True,
               capture_output=True,
              )

実行したコマンドがエラーだったら CalledProcessError の例外を発せさせる(check=True)。

try:
    p = subprocess.run(['ls', 'unknown'],
                       encoding='utf-8',
                       universal_newlines=True,
                       capture_output=True,
                       check=True,
                       )
except subprocess.CalledProcessError as e:
    print(e.returncode)  # 2
    print(e.cmd)  # ['ls', 'unknown']
    print(e.output, end='')  # ''
    print(e.stderr, end='') # ls: cannot access 'unknown': No such file or directory
else:
    print(p.stdout, end='')

パイプ

subprocess.run を使って shell=True することなくコマンドをパイプする。

import subprocess

def command_pipe(*args, input=None):
    proc = []
    for i, command in enumerate(args):
        if i > 0:
            # 前の実行結果を次に渡す
            input = proc[-1].stdout

        p = subprocess.run(command,
                           # input が None の場合は stdin に subprocess.PIPE が設定される
                           input=input,
                           # stdout と stderr に subprocess.PIPE を設定し、
                           # p.stdout, p.stderr にコマンドの実行結果を入れる
                           capture_output=True,
                           )
        proc.append(p)
        if p.stderr != b'':
            break

    return proc[-1]

p = command_pipe(['printf', 'foo\nbar\nbaz\n'], ['grep', 'foo'])
print(p.stdout.decode('utf-8'), end='') # foo

キーワード引数の input は文字列を受け取るために使う

s = """foo
bar
baz
"""

p = command_pipe(['grep', 'foo'], input=s)
print(p.stdout.decode('utf-8'), end='') # foo

コマンドラインでパイプやリダイレクトから何かを受け取る場合は input=sys.stdin.read() とかする必要はなくて input=None だったら内部的に stdin=subprocess.PIPE がセットされるので単に cat hello.txt | python main.py とか python main.py < hello.txt とすれば値が受け取れる。

p = command_pipe(['grep', 'foo'])
print(p.stdout.decode('utf-8'), end='')

$ printf 'foo\nbar\nbaz\n' | python main.py
foo

subprocess.Popen

Popen は非同期なので時間が掛かるコマンドを実行している間に何らかの処理ができる。

import subprocess
import time

p = subprocess.Popen(['sleep', '10'])

# Popenがコマンドを実行してる間に何かする
for i in range(1, 6):
    print(i)
    time.sleep(1)

p.wait()

subprocess.run 内部で Popen は以下のように使われてる。
https://github.com/python/cpython/blob/3.10/Lib/subprocess.py#L460

    with Popen(*popenargs, **kwargs) as process:
        try:
            stdout, stderr = process.communicate(input, timeout=timeout)
        except TimeoutExpired as exc:
            process.kill()
            if _mswindows:
                # Windows accumulates the output in a single blocking
                # read() call run on child threads, with the timeout
                # being done in a join() on those threads.  communicate()
                # _after_ kill() is required to collect that and add it
                # to the exception.
                exc.stdout, exc.stderr = process.communicate()
            else:
                # POSIX _communicate already populated the output so
                # far into the TimeoutExpired exception.
                process.wait()
            raise
        except:  # Including KeyboardInterrupt, communicate handled that.
            process.kill()
            # We don't call process.wait() as .__exit__ does that for us.
            raise
        retcode = process.poll()
        if check and retcode:
            raise CalledProcessError(retcode, process.args,
                                     output=stdout, stderr=stderr)
    return CompletedProcess(process.args, retcode, stdout, stderr)

vim を実行して親プロセスが終了してもいけるのかと思って色々ためしたけどダメだった。go言語だと syscall.Exec("vim", []string{"vim", "foo.txt"}, os.Environ()) とかでできた気がする。

import subprocess
import os
import sys

subprocess.Popen(['vim', 'foo.txt'], env=os.environ, shell=True, start_new_session=True)
sys.exit(0)