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)