シェルコマンドをPythonから実行して結果をリアルタイムに標準出力に出す方法

概要

Python の subprocess を使うことでシェルコマンドを実行可能だが、実行結果をリアルタイムに標準出力へとパイプする方法（かつエラーハンドリングが正しく行われる方法）がわかりづらいのでメモしておく。

ちなみに subprocess.run ではリアルタイムなパイプはできなかった。

方法

import subprocess
import sys


def run_shell(cmd):
    print(f'cmd: {cmd}')
    with subprocess.Popen(cmd, encoding='UTF-8', stdout=subprocess.PIPE, stderr=subprocess.PIPE) as p:
        for line in p.stdout:
            sys.stdout.write(line)
        p.wait()
        for line in p.stderr:
            sys.stdout.write(line)
        print(f'return: {p.returncode}')
        if p.returncode:
            raise Exception(f'Error! {cmd}')

• エラー出力もリアルタイムに出したいなら、stderr=subprocess.PIPE を stderr=subprocess.STDOUT とするのが手軽。

asyncio版

import asyncio
import sys

async def run_shell(cmd):
    print(f'cmd: {cmd}')
    process = await asyncio.create_subprocess_shell(
        cmd,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.STDOUT,
    )

    async for line in process.stdout:
        sys.stdout.write(line.decode('UTF-8'))
    
    returncode = await process.wait()
    
    print(f'return: {returncode}')
    if returncode:
        raise Exception(f'Error! {cmd}')

# asyncio.run(run_shell('your-command-here'))