Python の subprocess を理解する

概要

python から外部コマンドを実行したいケースは時々ありますね。
外部コマンドを実行して、標準出力、標準エラーをリアルタイムにリレーしたい。

そのための機能として subprocess モジュールがあります。
ただ、そのまま使うとどうも痒いところに手がとどかないし、
公式ドキュメント読んでもよくわからないところが多々あります。

ここでは subprocess モジュールを使って
（ときには asyncio.create_subprocess_shell を使って）
標準出力、標準エラーを適切にリレーする方法を解説します。

基本的な使い方

import subprocess
result = subprocess.run(['ls', '-l'])
print('return code:', result.returncode)  # 0なら成功

第1引数の ['ls', '-l'] は実行するコマンドと引数のリストです。
通常なら 'ls -l' と文字列で渡すところ、引数ごとに区切ることでセキュアになります。
たとえば、 input = '* @|@' の場合で ['echo', input] のようにユーザ入力に含める場合でも勝手に * が展開されたり、| が解釈されたりしなくなります。内部的にも若干早いです。

標準出力と標準エラーはそのまま実行環境の標準出力と標準エラーに出力されます。

shell: コマンドを配列ではなくシェルの文字列として渡したい

result = subprocess.run('ls -l *.txt', shell=True)
result = subprocess.run('echo "ls -l *.txt"', shell=True)

shell=True で文字列で渡すことができますが、エスケープは自己責任になります。
内部的にはシェルが１段階起動することになるので若干のオーバーヘッドが生じます。

注意: エスケープの仕様が実はややこしい

エスケープの仕様は OS ごとに異なります。

• Windows: cmd.exe のエスケープルールに従います。
• Mac/Linux: bash のエスケープルールに従います。

たとえば下記の方法で Windows/Mac 両方でエスケープできます。
（完全網羅はしていないのでユーザ入力を渡すならもっと精査が必要です。ユーザ入力を渡すなら配列で渡すのが絶対に良いです。）

def escape_cmd(cmd: str) -> str:
    """
    Windows なら Windows cmd.exe 用に特殊文字をエスケープする。
    Mac なら Linux 用にエスケープする。
    """
    if os.name == 'darwin':
        return shlex.quote(cmd)  # "" で囲まれたエスケープ済の文字列を返す
    else:
        replacements = {
            '^': '^^',
            '&': '^&',
            '|': '^|',
            '<': '^<',
            '>': '^>',
            '(': '^(',
            ')': '^)',
            '!': '^!',
            '%': '^%',
        }
        result = ''
        for ch in cmd:
            result += replacements.get(ch, ch)
        return f'"{result}"'

capture_output: 標準出力と標準エラーを取得したい

capture_output

result = subprocess.run(['ls', '-l'], capture_output=True, text=True)
print('return code:', result.returncode)  # 0なら成功
print('stdout:', result.stdout)           # 標準出力
print('stderr:', result.stderr)           # 標準エラー

capture_output=True は、標準出力と標準エラーをキャプチャするオプションです。キャプチャというのは、標準出力/標準エラーに出すのではなく、変数として取得するという意味です。
戻り値 の result でそれぞれ取得できるようになります。
ただし、この場合、標準出力も標準エラーもコンソールには出力されず、すべてが終わってからまとめてプロパティとして取れるようになります。

text=True は標準出力/標準エラーをバイト列ではなく文字列として取得したい場合に指定します。
つけなければバイト列として取得できるので zipfile モジュールに渡すなどの用途で使えます。この場合、標準エラーもバイト列になる点に注意が必要です。
標準出力だけ、という指定はできません。

標準エラーをリアルタイムにコンソールに出しつつ標準出力を取りたい

with subprocess.Popen(
        ['ls', '-l'],
        stdout=subprocess.PIPE,
        # text=True,  が無い
) as proc:
    stdout, _ = proc.communicate()
    print('return code:', proc.returncode)  # 0なら成功
    print('stdout:', stdout)                # 標準出力(ただしバイト列)

PIPE を使います。PIPE を扱うには低レイヤーのモジュール Popen を使う必要があります。
communicate() は PIPE した内容をすべて読み取るメソッドです。
PIPE の内容は全て読み取らないと処理が永久に止まってしまう（デッドロックになる）ので注意が必要ですが、communicate() は確実に読み取ってくれます。

標準エラーや標準出力をリアルタイムにコンソールに出しつつ取得もしたい

with subprocess.Popen(
        ['ls', '-l'],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True,
        bufsize=1,  # 行単位でバッファリング
) as proc:
    stdout = []
    stderr = []

    def _read_stream(in_stream: IO[AnyStr], out_stream: TextIO, out_list: list[str]):
        for line in in_stream:
            line = line.rstrip()
            print(line, file=out_stream)  # リアルタイムに標準エラーに出力
            out_list.append(line)

    thread_err = threading.Thread(target=_read_stream, args=(proc.stderr, sys.stderr, stderr))
    thread_err.start()
    _read_stream(proc.stdout, sys.stdout, stdout)
    thread_err.join()
    return_code = proc.wait()
    print('return code:', proc.returncode)  # 0なら成功
    print('stdout:', '\n'.join(stdout))     # 標準出力
    print('stderr:', '\n'.join(stderr))     # 標準エラー

communicate() では全て取得するまで止まってしまうので、スレッドによる読み取りが必要になります。

標準エラーはテキストで取得したいが、標準出力はバイト列で取得したい

with subprocess.Popen(
        ['ls', '-l'],
        stdout=subprocess.PIPE,
        # text=True, は指定しません
) as proc:
    stdout, _ = proc.communicate()
    print('return code:', proc.returncode)  # 0なら成功
    print('stdout:', stdout)     # 標準出力(バイト列)

stderr に対して何も指定しなければ、何の手も入れずにそのままコンソールに出力されます（結果としてテキストで表示されます）。
text を指定しなければ、標準出力はバイトで取れますのでこれで解決です。

cwd: ワークディレクトリを指定したい

result = subprocess.run(['ls', '-l'], cwd='/tmp')

cwd を省略したなら親プロセスのカレントディレクトリにいるものと見なされます。

env: 環境変数を指定したい（親プロセスの環境変数は引き継ぎたくない）

result = subprocess.run(['ls', '-l'], env={
    'XXX': 'xxx',
})

env で環境変数を追加できます。
env を指定しないなら親プロセスの環境変数が引き継がれますが、指定した場合は引き継がれず、完全に置き換えられます。
env={} とすれば親プロセスの環境変数を明示的に引き継がないことを指定できます。

env: 環境変数を指定したい（親プロセスの環境変数は引き継ぎたい）

result = subprocess.run(['ls', '-l'], env={
    **os.environ.copy(),
    'XXX': 'xxx',
})

**os.environ.copy() を入れることで親プロセスの環境変数を引き継いだうえで変更/追加できます。

asyncio.create_subprocess_shell: 非同期で外部コマンドを実行したい

process = await asyncio.create_subprocess_shell(
    ['ls', '-l'],
    stdout=asyncio.subprocess.PIPE,
    stderr=asyncio.subprocess.PIPE,
    cwd=cwd,
    env={
        **os.environ.copy(),
        'XXX': 'xxx',
    },
)
    stdout = []
    stderr = []
    
    async def read_stdout(reader: StreamReader, writer: TextIO, output: list[str]):
        async for line in reader:
            msg = line.decode('utf-8').rstrip()
            writer.write(msg + '\n')
            output.append(msg)
    
    # stdout, stderr を同時にリアルタイムで読む
    await asyncio.gather(
        read_stdout(process.stderr, sys.stderr, stderr),
        read_stdout(process.stdout, sys.stdout, stdout),
    )
    return_code = await process.wait()
    print('return code:', proc.returncode)  # 0なら成功

基本的な引数の考え方は subprocess と同じです。

まとめ

以上、いかがでしょうか。
簡単ですね。

こういう場合どうするの？ などコメント頂ければ随時追記していこうと思います。