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の状態毎のテストを書いたりしようとするとまたもう少し考えることが増えたりもするのだけれど)

pytestのpluginどこまで使うか

はじめに

pytestのpluginをどこまで使うかを考えるのがめんどくさくなったりした。

個人的にはそれぞれのタイミングで必要になったらはじめて手に取るべきで、あまり最初から導入するようなpluginは思いつかなかった。必要になったタイミングで追加して手に馴染むものがあれば使っていけば良いのでは位のスタンス。

一方で、デフォルトに乗るというか、大衆に従うというか、無難に使うということを考えた時に、どこまでのpytest pluginを利用するべきなんだろうかも気になったりした。

方法

例えば、以下の様に考えてみると、何が無難なパッケージかを判断できるかもしれない。

  1. pip search pytest で検索できるpackageの一覧を取り出す
  2. それぞれのpackageをdownloads数順に良い感じに並べる
  3. (追加で基準となりそうな馴染み深いパッケージも加えておく)
  4. ソートした後に、新たに追加された馴染み深いpackageを上回っているものが無難なパッケージ

ということで

pip search pytest で検索できるpackageの一覧を取り出す

$ pip search pytest | grep -v '^ ' | wc
    100     864    7127

100件くらい。どうにかなりそう。すごくびっくりするほど多いわけではない。

$ pip search pytest | grep -v '^ ' | head -n 3
pytest (4.5.0)                 - pytest: simple powerful testing with Python
pytest-httpserver (0.3.0)      - pytest-httpserver is a httpserver for pytest
pytest-bdd (3.1.0)             - BDD for pytest

それぞれのpackageをdownloads数順に良い感じに並べる

pypistatsでとりあえずてきとうに集める。あんまりたくさんAPIを呼ぶのはマナーが悪いけれど、100件くらいならまぁ良いでしょうということで(pypistatsについては以前書いた)。

$ pypistats recent pytest -f json
{"data": {"last_day": 456962, "last_month": 8510196, "last_week": 2674612}, "package": "pytest", "type": "recent_downloads"}

全部集める。

$ pip search pytest | grep -v '^ ' | cut -d " " -f 1 | xargs -I{} pypistats recent {} -f json | tee stats.json

(追加で基準となりそうな馴染み深いパッケージも加えておく)

なじみのある名前のパッケージも追加しておく

$ pypistats recent flake8 -f json >> stats.json
$ pypistats recent pandas -f json >> stats.json
$ pypistats recent flask -f json >> stats.json
$ pypistats recent arrow -f json >> stats.json
$ pypistats recent twine -f json >> stats.json
$ pypistats recent marshmallow -f json >> stats.json
$ pypistats recent aiohttp -f json >> stats.json

ソートした後に、新たに追加された馴染み深いpackageを上回っているものが無難なパッケージ

てきとうにsortしてあげる。そしてjsonからmarkdownにしてあげる。

$ jqfpy --slurp 'data = [{"package": d["package"], **d["data"]} for d in get()]; sorted(data, ke
y=lambda d: d["last_month"] + 4 * d["last_week"], reverse=True)' stats.json | dictknife cat -i json -o md | tee sorted.md

結論はこんな感じになった。

  • pytest-covくらいで良いのでは

もしかしたら以下のパッケージくらいは検討の余地があるかもしれない。

  • pytest-django (djangoのappを使っている場合は)
  • pytest-sugar
  • pytest-asyncio
  • pytest-repeat
  • pytest-aiohttp (aiohttpを使っている場合には)

それぞれのダウンロード数(直近のもの)

package last_day last_month last_week
pandas 426686 9451489 3018160
pytest 456962 8510196 2674612
aiohttp 618024 4751605 3223056
flask 376027 7474443 2322232
pytest-cov 150069 2783678 961471
flake8 145502 3023160 868458
arrow 64539 1316746 368395
marshmallow 45679 901455 257934
pytest-django 25164 478969 137959
twine 16439 353383 102052
pytest-sugar 7772 171525 48550
pytest-asyncio 7939 149570 44024
pytest-pep8 3331 79564 20891
pytest-repeat 2994 63269 17581
pytest-aiohttp 3107 56605 16705
pytest-selenium 3083 56677 15271
allure-pytest 2198 39679 12461
pytest-factoryboy 1688 32099 9805
pytest-bdd 1374 31177 8241
pytest-cover 787 13670 4348
pytest-sanic 896 13242 4454
pytest-doctestplus 508 12967 3817
pytest-salt 537 11821 3504
pytest-splinter 553 8464 2800
pytest-describe 490 5964 2388
pytest-faker 394 7393 1922
pytest-csv 212 6137 1600
pytest-pycodestyle 374 5050 1774
pytest-pudb 324 5126 1697
pytest-codestyle 244 4268 1350
pytest-trio 269 4342 1269
pytest-runfailed 171 3944 1070
pytest-xpara 165 3724 1008
behave-pytest 212 3189 893
pytest-regtest 127 3104 826
pytest-libfaketime 70 2125 626
pytest-ethereum 20 1873 675
pytest-invenio 153 1778 653
pytest-localstack 44 1966 410
pytest-mongodb 102 1668 433
pytest-testdox 31 1676 391
pytest-firefox 98 1487 419
pytest-grpc 104 1007 413
pybuilder-pytest 19 1034 342
pytest-docstyle 41 1152 278
pytest-testconfig 68 796 357
pytest-warnings 36 952 228
pytest-fxa 47 783 243
pytest-pydocstyle 57 724 251
pytest-httpserver 54 618 232
hypothesis-pytest 42 709 170
pytest-sqlalchemy 40 497 201
pytest-expecter 16 613 166
pygments-pytest 11 657 143
pytest-httpretty 23 399 131
pytest-fixtures 18 403 108
pytest-smartcov 47 342 80
pytest-monkeyplus 6 198 100
pytest-dynamodb 8 281 77
pytest-falcon 17 292 73
pytest-bandit 8 290 50
pytest-monkeytype 3 240 55
pytest-toolbox 1 245 45
pytest-pdb 5 223 41
pytest-annotate 1 290 13
pytest-slack 2 186 37
pytest-diff 2 209 31
pytest-curio 0 163 35
pytest-neo 3 213 19
pytest-rage 10 127 38
pytest-reqs 1 190 21
pytest-excel 4 131 33
pytest-trepan 7 150 26
pytest-sourceorder 18 129 31
pytest-jest 3 102 34
pytest-bench 1 151 21
pytest-fauxfactory 0 107 31
pytest-reana 0 202 6
pytest-flake8dir 0 122 21
pytest-race 5 92 27
pytest-docs 2 124 11
pytest-airflow 0 115 12
pytest-ponyorm 1 113 12
pytest-ngsfixtures 0 145 4
pytest-pyq 2 77 20
pytest-apistellar 0 112 11
pytest-orchestration 1 102 13
pytest-statsd 3 97 11
pytest-assertutil 0 97 8
pytest-datatest 2 66 10
pytest-diamond 2 45 11
pytest-tesults 0 56 8
pytest-rt 0 47 9
pytest-pytestrail 0 50 8
pytest-zafira 0 52 6
pytest-scenario 1 49 6
pytest-cram 0 42 5
pytest-ok 1 36 6
geoffrey-pytest 0 36 6
pytest-pact 0 33 6
pytest-growl 1 34 5
pytest-rethinkdb 0 35 4
pytest-cricri 0 31 4
pytest-nginx 0 29 3
pytest-symbols 0 23 4
pytest-redmine 0 21 3

memo

以下をinstallする必要があるかも

$ pip install pypistats jqfpy dictknife