閉じる

Pythonのsubprocessで標準出力を取得する方法:2026年時点の最適な実装パターン

Pythonで外部プログラムを実行し、その結果を取得する処理は、システムの自動化やツール開発において極めて頻繁に利用される機能です。

2026年現在のPython開発においては、かつての os.systemcommands モジュールを使用することは推奨されず、subprocessモジュールを用いるのが標準となっています。

本記事では、標準出力を取得するための最も効率的で安全な実装パターンについて、最新のベストプラクティスを交えて詳しく解説します。

subprocess.runを使用した標準出力の取得

現代のPythonにおいて、外部コマンドを実行するための最も推奨される方法は subprocess.run() 関数を利用することです。

この関数はPython 3.5で導入されて以来、改良が重ねられ、現在ではほとんどのユースケースをカバーできる万能なAPIとして定着しています。

基本的な実装パターン

もっともシンプルに標準出力を取得するには、capture_output=Truetext=True という2つの引数を指定します。

Python
import subprocess

# lsコマンド(Windowsの場合は'dir')を実行して出力を取得
# capture_output=True: 標準出力と標準エラー出力をキャプチャする
# text=True: 出力を文字列(str)として取得する(指定しない場合はbytes型)
result = subprocess.run(["ls", "-l"], capture_output=True, text=True)

# 取得した標準出力を表示
print(result.stdout)

実行結果は以下のようになります(環境によって内容は異なります)。

text
total 8
-rw-r--r--  1 user  staff  128 May  4 10:00 main.py
-rw-r--r--  1 user  staff  512 May  4 10:05 script.py

引数の詳細解説

上記のコードで使用した引数には、それぞれ重要な役割があります。

  1. リスト形式のコマンド指定: コマンドとその引数はリスト形式で渡すことが推奨されます。これにより、OSのシェルによる解釈を経由しないため、シェル注入(Shell Injection)攻撃のリスクを最小限に抑えることができます。
  2. capture_output=True: これを指定することで、実行したプロセスの stdout(標準出力)と stderr(標準エラー出力)の両方が、返り値である CompletedProcess オブジェクトに格納されます。
  3. text=True: このオプションを有効にすると、Pythonは取得したバイナリデータをデフォルトのエンコーディング(通常は UTF-8)でデコードし、文字列として扱えるようにします。以前のバージョンでは universal_newlines=True と記述されていましたが、現在は text=True と書くのが一般的です。

標準出力と標準エラー出力を個別に扱う

実務的なスクリプトでは、正常な出力とエラーメッセージを区別して処理する必要があります。

subprocess.run() の返り値である CompletedProcess は、これらを個別の属性として保持しています。

stdoutとstderrの分離

以下の例では、存在しないファイルを指定してエラーを発生させ、その出力を確認します。

Python
import subprocess

# 存在しないファイルをリストしようとするコマンド
result = subprocess.run(["ls", "non_existent_file"], capture_output=True, text=True)

# 標準出力(空のはず)
print(f"STDOUT: {result.stdout}")

# 標準エラー出力(エラーメッセージが含まれる)
print(f"STDERR: {result.stderr}")

# 終了コードの確認(0以外は異常終了)
print(f"Return Code: {result.returncode}")
実行結果
STDOUT: 
STDERR: ls: non_existent_file: No such file or directory
Return Code: 1

エラー発生時に例外をスローする

コマンドが失敗した(終了コードが0以外だった)場合に自動的に例外を発生させたい場合は、check=True 引数を追加します。

これにより、try-except ブロックでエラーハンドリングを記述しやすくなります。

Python
import subprocess

try:
    # check=Trueにより、エラー時にCalledProcessErrorが発生する
    result = subprocess.run(
        ["ls", "non_existent_file"], 
        capture_output=True, 
        text=True, 
        check=True
    )
except subprocess.CalledProcessError as e:
    print(f"コマンドの実行に失敗しました。終了コード: {e.returncode}")
    print(f"エラー内容: {e.stderr}")

リアルタイムで標準出力を取得する方法

subprocess.run() は非常に便利ですが、一つ欠点があります。

それは、外部プロセスの実行が完全に終了するまで制御が戻らないという点です。

実行に数分かかるようなプログラムの場合、進捗をリアルタイムで確認したいことがあります。

その場合は、より低レベルなクラスである subprocess.Popen を使用します。

Popenによるストリーミング処理

Popen を使うと、プロセスの実行中にその出力を1行ずつ読み取ることが可能です。

Python
import subprocess

# 長時間かかる処理をシミュレートするコマンドの例
# 例として、5秒間1秒ごとに数値を出力する
cmd = ["python3", "-c", "import time; [print(i) or time.sleep(1) for i in range(5)]"]

process = subprocess.Popen(
    cmd,
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    text=True
)

# 実行中の出力を1行ずつ読み取る
print("実行開始...")
while True:
    line = process.stdout.readline()
    if not line and process.poll() is not None:
        break
    if line:
        print(f"リアルタイム受信: {line.strip()}")

# 最終的な終了ステータスを確認
return_code = process.wait()
print(f"プロセスが終了しました。終了コード: {return_code}")

実行結果は、約1秒ごとに1行ずつ表示されます。

text
実行開始...
リアルタイム受信: 0
リアルタイム受信: 1
リアルタイム受信: 2
リアルタイム受信: 3
リアルタイム受信: 4
プロセスが終了しました。終了コード: 0

注意点:バッファリングの影響

リアルタイム取得を行う際、実行するプログラム側の実装によっては バッファリング が原因で出力が遅延することがあります。

Pythonスクリプトを呼び出す場合は、-u オプションを付けるか、flush=True を使ってバッファを強制的に出力させる必要があります。

2026年における最適な実装パターン:セキュリティと堅牢性

プログラミング環境が進化するにつれ、セキュリティとコードの可読性に対する要求は高まっています。

subprocessを使用する際の最新のガイドラインをまとめます。

shell=Trueを回避する

shell=True を指定すると、文字列をそのままシェルに渡せますが、これは極めて危険です。

ユーザー入力を含む文字列をそのまま渡すと、任意のコマンドを実行される脆弱性に繋がります。

項目推奨される方法 (List形式)非推奨の方法 (String形式 + shell=True)
安全性高い(インジェクションを防ぐ)低い(脆弱性の原因)
可読性明確(引数が分離されている)複雑なエスケープが必要な場合がある
パフォーマンスシェルを起動しないため高速シェルの起動コストがかかる

タイムアウトの設定

外部プロセスが何らかの理由でフリーズした場合、メインのPythonプログラムも停止してしまいます。

これを防ぐために、timeout 引数を常に検討すべきです。

Python
import subprocess

try:
    # 2秒以内に終了しない場合はTimeoutExpired例外を出す
    result = subprocess.run(["sleep", "10"], capture_output=True, timeout=2)
except subprocess.TimeoutExpired as e:
    print("処理がタイムアウトしました。")
    # タイムアウトまでに取得できた出力を確認することも可能
    if e.stdout:
        print(f"途中経過: {e.stdout.decode()}")

非同期処理 (asyncio) との連携

現代のPython開発、特にWebアプリケーションや大量のI/Oを扱う環境では、asyncio との組み合わせが一般的です。

標準の subprocess はブロッキング(処理を止める)関数のため、非同期環境では asyncio.create_subprocess_exec を使用します。

Python
import asyncio

async def run_command_async():
    # 非同期で外部プロセスを実行
    process = await asyncio.create_subprocess_exec(
        "ls", "-l",
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE
    )

    # 標準出力を非同期で読み取る
    stdout, stderr = await process.communicate()

    if stdout:
        print(f"[stdout]\n{stdout.decode()}")
    if stderr:
        print(f"[stderr]\n{stderr.decode()}")

# Python 3.7+ の実行方法
asyncio.run(run_command_async())

この方法を用いることで、外部プロセスの完了を待っている間も、他の非同期タスクを並行して実行することが可能になります。

これはスケーラビリティが求められる現代のバックエンド開発において非常に重要なテクニックです。

文字エンコーディングのトラブルシューティング

標準出力を取得する際に最も多いトラブルが、文字化けやデコードエラーです。

特にWindows環境での日本語処理には注意が必要です。

適切なエンコーディングの指定

text=True を使用すると環境のデフォルトエンコーディングが使われますが、出力されるデータのエンコーディングが分かっている場合は明示的に指定するのが安全です。

Python
# Windowsのコマンドプロンプト(CP932)の結果を取得する場合
result = subprocess.run(["ipconfig"], capture_output=True, text=True, encoding="cp932")

また、予期しない文字が含まれている場合にエラーでプログラムを止めたくない場合は、errors="replace" を指定して、不正な文字を置換するように設定することも可能です。

まとめ

2026年現在、Pythonで標準出力を取得するための最適解は、用途に応じて以下の2つに集約されます。

  • 単発の実行: subprocess.run(..., capture_output=True, text=True) を使用し、安全に結果を一括取得する。
  • 進捗の監視が必要な場合: subprocess.Popen を使用し、stdout.readline() で1行ずつ処理する。

いずれの場合も、セキュリティのために shell=True を避け、リスト形式でコマンドを渡す習慣をつけることが重要です。

また、タイムアウトの設定や適切なエンコーディングの指定など、実行環境に依存する部分を丁寧にケアすることで、堅牢なスクリプトを構築することができます。

これらのパターンをマスターすることで、Pythonを通じたシステム操作やツール連携のクオリティは格段に向上するでしょう。

プロジェクトの要件に合わせて、最適な手法を選択してください。

クラウドSSLサイトシールは安心の証です。

URLをコピーしました!