pythonのsubprocessを使ったあれこれあるいはsubprocess.runのすすめ

はじめに

pythonでsubprocessを使ってあれこれする方法をイディオム的に覚えておくと便利なのでまとめておく。

できること（やろうとしていること）は以下

stdoutを文字列として受け取る
文字列をstdinとして渡す
stdoutをiteratorとして受け取る

(その前に) subprocess.run

その前に何はともあれ注意事項を。

なるべく使えるなら subprocess.run() を使ったほうが良い。つまり、はじめに run() で書くことを考えてダメそうだったら他の方法を考えるという手順で付き合う感じ。

https://docs.python.org/ja/3/library/subprocess.html#using-the-subprocess-module より

サブプロセスを起動するために推奨される方法は、すべての用法を扱える run() 関数を使用することです。より高度な用法では下層の Popen インターフェースを直接使用することもできます。

stdoutを文字列として受け取る

stdoutを文字列として受け取るはそのまま返ってきたCompletedProcessのstdoutが結果を保持している。

run()の引数にtext=Trueを渡してあげると、bytesではなくstrとして返してくれるので便利。

00tostring.py

import subprocess

p = subprocess.run(["ls"], stdout=subprocess.PIPE, text=True)
print(p.stdout.strip())

実行結果

# lsの結果なので変わる
$ python 00*.py
00tostring.py
01provide_stdin.py
readme.md

文字列をstdinとして渡す

文字列をstdinとして渡す方法もsubprocess.run()が対応している。runの引数のinputに文字列を渡してあげれば良い。

01provide_stdin.py

import subprocess

input_text = """
aaa
bbb
ccc
""".strip()

subprocess.run(["sed", "s/^/@@/g"], input=input_text, text=True)

実行結果

$ python 01*.py
@@aaa
@@bbb
@@ccc

stdoutをiteratorとして受け取る

stdoutをiteratorとして受け取る方法はちょっとトリッキーかもしれない。iter()で空文字列が返されたら終了することにする。手軽な処理ならこれくらいで十分な気がする。

注意点としては呼び出される側の処理も内部のバッファに溜め込まずにflushしてあげるように描く必要があるかもしれない(例えば、pythonなら -u 付きで呼び出す。あるいは sys.stdout.flush() を都度呼び出すなど)。

02as_iterator.py

import time
import subprocess


cmd = [
    "python",
    "-u",
    "-c",
    "import time; [print(i) or time.sleep(0.4) for i in range(10)]",
]
with subprocess.Popen(cmd, text=True, stdout=subprocess.PIPE) as p:
    for line in iter(p.stdout.readline, ""):
        print("* ", time.time(), line.rstrip())

実行結果

python 02*.py
*  1558516806.159693 0
*  1558516806.5610576 1
*  1558516806.9615686 2
*  1558516807.362776 3
*  1558516807.7639756 4
*  1558516808.1644542 5
*  1558516808.565165 6
*  1558516808.9661117 7
*  1558516809.3671272 8
*  1558516809.7683222 9

(もう少し本気を出す(?)ならthreadingでThreadを立ち上げてqueueに渡しqueueを消費するgeneratorを書くなどするかもしれない。あるいはiteratorとして作るのを諦めてgeneratorとsendを使って通信だとか)

どういう時にsubprocess.run()で十分か?

どういう時にsubprocess.run()で十分か?というと概ね同期的な処理を志向しているかどうかで判定できる（と思う）。

同期的な処理で良いということは、つまり文字列を受け取って文字列を返す関数と見做してしまって良いかということ（引数として文字列を受け取らない場合もあるかもしれない）。subprocessの実行中にメインの処理はただただ待つだけという形なら同期的な処理を施行していると言えそう。

そういう意味では例えば最後のiteratorとして取り出す処理あたりはちょっと怪しい(のでrun()では表せない)。

run()による抽象の嬉しさ

run()による抽象の何が嬉しいかと言うと、テストの時にとてもうれしくなる。というのも単なる文字列を受け取って(あるいは受け取らなかったり)文字列を返すただの関数としてみなせるので。

use_input :: Callable[[str, str], None]
use_output  :: Callable[[str], str]
use_both :: Callable[[str, str], str]

こういう形に、

def use_input(cmd, input) -> None:
    do_something()  # side effect?


def use_output(cmd) -> str:
    # do sometihg
    return "<result>"

そしてそれらを使う関数たちは、例えば高階関数の形にしてあげればテスト時に引数としてfakeの実装を渡せば良いし、あるいは直接実行される関数であっても最悪unittest.mockでpatchしてあげれば済む。

このときのmockで差し替える実装もまた単純な(文字列を返すあるいは受け取る)関数で良いので便利。

高階関数の例(まぁinputの方でこのように書く事はあまりない気がするけれど)

def do_something(*, use_input=None):
    # 場合によってはdefault実装を用意しておくこともあるかもしれない
    use_input = use_input or default_use_input

    input_text = do_something()
    use_input(input_text)

テストの例

class Test(unittest.TestCase):
    def test_it(self):
        from xxx import do_something

        box = []
        def _use_input(text):
            box.append(text)

        do_something(use_input=_use_input)

        # assertion (このあたりは各自お好みの方法で)
        self.assertEqual(len(box), 1)

この様になっていなかったとしても、

def do_something2():
    input_text = do_something()
    use_input(input_text)

テストで差し替える仮の実装を用意するのが手軽（何かをシミュレートすることも無いし、何かと通信する必要もない）。

import unittest.mock as mock


class Test(unittest.TestCase):
    def test_it(self):
        box = []

        def _use_input(text):
            box.append(text)

        with mock.patch("xxx.use_input") as m:
            m.side_effect = _use_input
            from xxx import do_something2
            do_something2()

        # assertion (このあたりは各自お好みの方法で)
        self.assertEqual(len(box), 1)

逆に言うとせめて以下の様な形で定義はされていて欲しいということかも。

def use_input(input_text):
    # その他色々な処理
    subprocess.run(cmd, input=input_text, ...)
    # その他色々な処理

これは出力の利用も同じ話。

（まぁprocessの状態毎のテストを書いたりしようとするとまたもう少し考えることが増えたりもするのだけれど）