🐍 Pythonのsubprocessモジュール徹底解説：外部コマンドを自在に操る

はじめに：subprocessモジュールとは？ 🤔

Pythonを使ってシステム管理タスクや他のプログラムとの連携を行う際、外部のコマンドやプロセスを実行したい場面は非常に多くあります。例えば、シェルのコマンドを実行してファイル操作を行ったり、別の言語で書かれたプログラムを実行してその結果を取得したり、といったケースです。

このような外部プロセスとの連携を実現するために、Pythonの標準ライブラリとして提供されているのが subprocess モジュールです。このモジュールを使うことで、新しいプロセスの生成、そのプロセスへの入力、出力の取得、終了ステータスの確認などを簡単かつ安全に行うことができます。

かつては os.system() や os.spawn*()、os.popen*() といった関数も利用されていましたが、これらは機能が限定的であったり、セキュリティ上の問題を抱えていたりしました。subprocess モジュールは、これらの旧来の方法を置き換え、より柔軟で強力なプロセス管理機能を提供するために導入されました（特にPython 2.4以降）。

このブログ記事では、subprocess モジュールの基本的な使い方から、標準入出力のリダイレクト、エラーハンドリング、パイプライン処理、非同期実行、そして重要なセキュリティに関する注意点まで、幅広く掘り下げて解説していきます。さあ、subprocess の世界を探検しましょう！ 🚀

基本的な使い方：subprocess.run() 関数 ✨

subprocess モジュールで最も推奨され、一般的に使われるのが subprocess.run() 関数です。これはPython 3.5で導入され、外部コマンドを実行し、その完了を待つためのシンプルかつ強力なインターフェースを提供します。

基本的な構文は以下のようになります。

import subprocess

# コマンドと引数をリストで指定
result = subprocess.run(['ls', '-l'], capture_output=True, text=True)

# 実行結果の確認
print("Return code:", result.returncode)
print("Stdout:", result.stdout)
print("Stderr:", result.stderr)
      

上記の例では、ls -l コマンドを実行しています。重要なポイントは以下の通りです。

• コマンドと引数: 実行したいコマンドとその引数は、文字列のリスト（またはタプル）として run() 関数の第一引数 (args) に渡すのが基本です。これは後述するセキュリティの観点からも推奨される方法です。
• capture_output=True: この引数を指定すると、コマンドの標準出力 (stdout) と標準エラー出力 (stderr) がキャプチャされ、戻り値のオブジェクトからアクセスできるようになります。指定しない場合、出力はPythonスクリプトを実行しているターミナルなどにそのまま表示され、結果オブジェクトには含まれません。
• text=True: Python 3.7以降で利用可能です。（それ以前は universal_newlines=True）。これを指定すると、stdin, stdout, stderr がテキストモードで扱われ、自動的にシステムのデフォルトエンコーディングでデコード・エンコードされます。結果として、result.stdout や result.stderr がバイト列 (bytes) ではなく文字列 (str) になります。これにより、テキストデータの扱いが非常に楽になります 😊。指定しない場合はバイト列として扱われます。
• 戻り値: run() 関数は CompletedProcess オブジェクトを返します。このオブジェクトには、実行されたコマンド、引数 (args)、終了コード (returncode)、キャプチャされた標準出力 (stdout)、標準エラー出力 (stderr) などが含まれます。
• 終了コード (returncode): コマンドが正常に終了した場合は通常 0 になります。エラーが発生した場合は 0 以外の値（通常は正の整数）が返されます。

run() 関数の主要な引数

run() 関数には多くの便利な引数があります。いくつか重要なものを紹介します。

標準入出力のリダイレクト 🔁

subprocess の強力な機能の一つが、子プロセスの標準入力 (stdin)、標準出力 (stdout)、標準エラー出力 (stderr) を柔軟に制御できる点です。これは run() 関数の stdin, stdout, stderr 引数を使って行います。

標準出力 (stdout) と標準エラー出力 (stderr) のキャプチャ

既に見たように、capture_output=True または stdout=subprocess.PIPE, stderr=subprocess.PIPE を指定することで、子プロセスの出力を取得できます。

import subprocess

# stdoutとstderrをキャプチャして文字列として取得
result = subprocess.run(['cat', 'non_existent_file.txt'], capture_output=True, text=True)

# stdoutは空のはず
print(f"Stdout: '{result.stdout}'")
# stderrにエラーメッセージが含まれる
print(f"Stderr: '{result.stderr}'")
print(f"Return code: {result.returncode}") # 0以外になるはず
      

標準入力 (stdin) へのデータ送信

子プロセスにデータを入力として渡したい場合は、run() 関数の input 引数を使用します。この場合、stdin=subprocess.PIPE を明示的に指定する必要はありません (input を使うと自動的に設定されます)。

input に渡すデータは、text=True の場合は文字列、text=False (デフォルト) の場合はバイト列である必要があります。

import subprocess

# grepコマンドに標準入力からデータを渡す
input_data = "line1\nline2 with keyword\nline3\nline4 with keyword too"
result = subprocess.run(
    ['grep', 'keyword'],
    input=input_data,
    capture_output=True,
    text=True # inputもtextとして扱われる
)

print("Grep Return code:", result.returncode)
print("Grep Stdout:\n", result.stdout)
print("Grep Stderr:", result.stderr)
      

出力のリダイレクト先

stdout や stderr には subprocess.PIPE 以外も指定できます。

• subprocess.DEVNULL: 出力を完全に破棄します。ログなどを抑制したい場合に便利です。
• 既存のファイルディスクリプタ: 例えば sys.stdout や sys.stderr を指定すると、親プロセスと同じ出力先に書き出されます（デフォルトの動作に近いですが、明示的に指定することも可能です）。
• ファイルオブジェクト: open() で開いたファイルオブジェクトを指定すると、そのファイルに出力が書き込まれます。

import subprocess
import sys

# stderrをファイルにリダイレクト
try:
    with open('error.log', 'w') as f_err:
        result = subprocess.run(
            ['cat', 'non_existent_file.txt'],
            stdout=subprocess.PIPE, # stdoutはキャプチャ
            stderr=f_err,          # stderrはファイルへ
            text=True
        )
    print("Command executed.")
    # error.logを確認するとエラーメッセージが書き込まれているはず
except FileNotFoundError:
    print("Error: 'cat' command not found.")
except Exception as e:
    print(f"An error occurred: {e}")

# stdoutを破棄し、stderrは親プロセスと同じにする
try:
    result_devnull = subprocess.run(
        ['ls', '/'],
        stdout=subprocess.DEVNULL, # 標準出力は表示しない
        stderr=sys.stderr,        # 標準エラーはターミナルに出力
        text=True
    )
    print(f"\nls command return code: {result_devnull.returncode}")
except FileNotFoundError:
    print("Error: 'ls' command not found.")
except Exception as e:
    print(f"An error occurred: {e}")
      

stdout と stderr のマージ

標準出力と標準エラー出力を区別せず、同じストリームで扱いたい場合があります。その場合は、stderr=subprocess.STDOUT を指定します。これにより、stderr が stdout と同じハンドルにリダイレクトされます。

import subprocess

# findコマンドで、見つかったファイル(stdout)と権限エラー(stderr)をマージしてキャプチャ
try:
    result = subprocess.run(
        ['find', '/', '-name', 'python*', '-print'], # '-print' はなくても良い場合が多い
        stdout=subprocess.PIPE, # stdoutをキャプチャ
        stderr=subprocess.STDOUT, # stderrをstdoutにマージ
        text=True
    )
    print("Find Return code:", result.returncode) # エラーがあっても0になることがあるので注意
    print("Combined Output:\n", result.stdout)
    # result.stderr は None になる
    print("Stderr:", result.stderr)
except FileNotFoundError:
    print("Error: 'find' command not found.")
except Exception as e:
    print(f"An error occurred: {e}")
      

エラーハンドリング 🚨

外部コマンドを実行する際には、様々なエラーが発生する可能性があります。コマンドが見つからない、実行権限がない、コマンド自体がエラーを返す、タイムアウトするなどです。これらのエラーに適切に対処することが重要です。

終了コードの確認

最も基本的なエラーチェックは、CompletedProcess オブジェクトの returncode 属性を確認することです。通常、0 は成功を意味し、それ以外の値はエラーを示します。

import subprocess

result = subprocess.run(['false']) # 'false' コマンドは常に非ゼロの終了コードを返す
if result.returncode != 0:
    print(f"Command failed with return code: {result.returncode}")
else:
    print("Command succeeded.")
      

check=True による例外発生

コマンドが失敗した場合（終了コードが 0 でない場合）に、自動的に例外を発生させたい場合は、run() 関数に check=True を指定します。これにより、CalledProcessError 例外が送出されます。

CalledProcessError 例外オブジェクトは、returncode, cmd, output (stdout), stderr 属性を持っており、エラーの詳細を確認できます。

import subprocess

try:
    # 存在しないファイルを指定してエラーを発生させる
    result = subprocess.run(['ls', 'non_existent_directory'], check=True, capture_output=True, text=True)
    print("Command succeeded (should not happen).")
    print("Stdout:", result.stdout)
except FileNotFoundError as e:
    print(f"Error: Command not found - {e}") # lsコマンド自体が見つからない場合
except subprocess.CalledProcessError as e:
    print(f"Command failed!")
    print(f"  Command: {e.cmd}")
    print(f"  Return code: {e.returncode}")
    print(f"  Stdout: {e.stdout}") # capture_output=Trueならキャプチャされている
    print(f"  Stderr: {e.stderr}") # capture_output=Trueならキャプチャされている
except Exception as e:
    print(f"An unexpected error occurred: {e}")
      

check=True は、コマンドが成功することを前提としている場合にコードを簡潔にするのに役立ちます。エラー処理を明示的に行いたい場合は、try...except ブロックで CalledProcessError を捕捉します。

コマンドが見つからない場合のエラー

実行しようとしたコマンド自体が存在しない場合、FileNotFoundError が発生します。これも try...except で捕捉する必要があります。

import subprocess

try:
    # 存在しないであろうコマンドを実行
    subprocess.run(['non_existent_command_blah_blah'], check=True)
except FileNotFoundError as e:
    print(f"Error: The command was not found: {e}")
except subprocess.CalledProcessError as e:
    print(f"Command failed with return code: {e.returncode}")
except Exception as e:
    print(f"An unexpected error occurred: {e}")
      

タイムアウト処理

外部コマンドが予期せず長時間実行され、プログラム全体が停止してしまうのを防ぐために、timeout 引数を使用できます。指定した秒数内にコマンドが終了しない場合、TimeoutExpired 例外が発生します。

import subprocess

try:
    # 5秒間スリープするコマンドを3秒のタイムアウトで実行
    result = subprocess.run(['sleep', '5'], timeout=3)
    print("Command finished within timeout (should not happen).")
except FileNotFoundError:
    print("Error: 'sleep' command not found.")
except subprocess.TimeoutExpired as e:
    print(f"Command timed out after {e.timeout} seconds!")
    print(f"  Command: {e.cmd}")
    # タイムアウトした場合でも、stdoutやstderrに部分的な出力が含まれている可能性がある
    print(f"  Stdout (partial): {e.stdout}")
    print(f"  Stderr (partial): {e.stderr}")
except Exception as e:
    print(f"An unexpected error occurred: {e}")
      

TimeoutExpired 例外が発生した場合、子プロセスは SIGKILL シグナルで強制終了されます。例外オブジェクトには、タイムアウト時点までにキャプチャされた stdout や stderr (バイト列) が含まれていることがあります。

パイプライン処理 🔗

シェルでは、あるコマンドの出力を別のコマンドの入力に繋ぐ「パイプライン」(|) がよく使われます。subprocess モジュールを使っても、同様の処理を実現できます。これには、主に subprocess.Popen クラスを使用します（run() でも可能ですが、Popen の方がより直接的です）。

Popen は run() と異なり、プロセスを開始した直後に制御を返し、プロセスの完了を待ちません（非同期的）。プロセスの完了を待ったり、入出力を処理したりするには、追加のメソッド呼び出しが必要です。

例として、ls -l の出力を grep python にパイプで繋ぐ処理を考えてみましょう。

import subprocess

try:
    # 第一のプロセス (ls -l) を開始
    p1 = subprocess.Popen(['ls', '-l'], stdout=subprocess.PIPE)

    # 第二のプロセス (grep python) を開始
    # stdinには第一のプロセスのstdoutを指定
    p2 = subprocess.Popen(['grep', 'python'], stdin=p1.stdout, stdout=subprocess.PIPE, text=True)

    # 第一のプロセスのstdoutを閉じる（p2がEOFを受け取れるように）
    # これを行わないと、p2がp1の終了を待ち続けてデッドロックする可能性がある
    if p1.stdout:
        p1.stdout.close()

    # 第二のプロセスの出力を取得
    # communicate() はプロセスの完了を待ち、stdoutとstderrを返す
    stdout, stderr = p2.communicate()

    print("Pipeline Output:")
    print(stdout)

    # 各プロセスの終了コードも確認できる (オプション)
    # communicate() の後でなければ wait() は不要なことが多い
    # p1_rc = p1.wait()
    p2_rc = p2.wait()
    print(f"\nls return code: {p1.returncode}") # Popenオブジェクトは完了後もreturncodeを持つ
    print(f"grep return code: {p2_rc}")

except FileNotFoundError as e:
    print(f"Error: Command not found - {e}")
except Exception as e:
    print(f"An error occurred: {e}")

      

この例のポイントは以下の通りです。

• Popen を使って各プロセスを開始します。
• 最初のプロセス (p1) の stdout を subprocess.PIPE に設定します。
• 次のプロセス (p2) の stdin に、前のプロセス (p1) の stdout オブジェクト (p1.stdout) を指定します。
• p2 を開始した後、p1.stdout.close() を呼び出すことが重要です。これにより、p1 が終了した際にパイプが閉じられ、p2 が入力の終わり (EOF) を検知できるようになります。これを忘れると、p2 が入力を待ち続けてデッドロックする可能性があります。
• 最後のプロセスの communicate() メソッドを呼び出して、最終的な出力を取得し、プロセスの完了を待ちます。communicate() は内部で必要な待機処理を行うため、通常は個別に wait() を呼ぶ必要はありません。

なお、subprocess.run() でも、input 引数を使ってパイプラインのような処理を模倣できますが、中間プロセスの制御やエラーハンドリングは Popen を使う方が柔軟です。

import subprocess

try:
    # ls -l の出力を取得
    p1_result = subprocess.run(['ls', '-l'], capture_output=True, text=True, check=True)

    # grep に ls の出力を input として渡す
    p2_result = subprocess.run(
        ['grep', 'python'],
        input=p1_result.stdout, # 前のコマンドのstdoutを渡す
        capture_output=True,
        text=True,
        check=True # grepが何も見つけられなくてもエラーにはしない場合はFalseにする
    )

    print("Pipeline Output (using run):")
    print(p2_result.stdout)

except FileNotFoundError as e:
    print(f"Error: Command not found - {e}")
except subprocess.CalledProcessError as e:
    print(f"Command failed in pipeline!")
    print(f"  Command: {e.cmd}")
    print(f"  Return code: {e.returncode}")
    print(f"  Stderr: {e.stderr}")
except Exception as e:
    print(f"An error occurred: {e}")
      

非同期処理：Popenクラスの活用 ⏳

subprocess.run() はコマンドが完了するまで待機する同期的な関数です。しかし、時間がかかる可能性のあるコマンドを実行しつつ、Pythonスクリプト側で他の処理を続けたい場合や、複数のプロセスを並行して実行したい場合があります。このような非同期的な処理には subprocess.Popen クラスを使用します。

Popen は run() と同様の引数を多く取りますが、インスタンス化するとすぐに子プロセスを開始し、制御を返します。

import subprocess
import time

print("Starting long-running process...")
# 5秒かかるプロセスを開始 (バックグラウンドで実行される)
process = subprocess.Popen(['sleep', '5'])
print(f"Process started with PID: {process.pid}. Continuing with other tasks...")

# 他の処理を実行
for i in range(3):
    print(f"Doing other work... ({i+1}/3)")
    time.sleep(1)

print("Checking if process has finished...")
# poll() はプロセスが終了していれば終了コード、していなければ None を返す (ブロックしない)
return_code = process.poll()
if return_code is None:
    print("Process is still running.")
else:
    print(f"Process finished with return code: {return_code}")

# プロセスの完了を待つ場合は wait() を使う
print("Waiting for process to complete...")
# wait() はプロセスが終了するまでブロックし、終了コードを返す
final_return_code = process.wait()
print(f"Process completed with return code: {final_return_code}")

# 再度 poll() を呼ぶと、終了コードが返る
print(f"Return code after wait(): {process.poll()}")
      

Popenオブジェクトのメソッド

Popen オブジェクトには、開始したプロセスを制御するためのメソッドが用意されています。

communicate() の注意点

communicate() は非常に便利ですが、注意点があります。このメソッドは、プロセスの stdout と stderr からすべての出力をメモリに読み込もうとします。そのため、非常に大量の出力を生成するプロセスに対して使用すると、メモリを大量に消費してしまう可能性があります。そのような場合は、Popen オブジェクトの stdout や stderr 属性（ファイルライクオブジェクト）を直接、少しずつ読み書きする高度な処理が必要になることがあります。

import subprocess

# 大量の出力を生成する例 (ここでは 'yes' コマンドを使用)
# 注意: 'yes' は無限に出力し続けるため、タイムアウトや手動停止が必要です
try:
    process = subprocess.Popen(['yes'], stdout=subprocess.PIPE, text=True)

    print(f"Started 'yes' process (PID: {process.pid}). Reading first 10 lines...")

    line_count = 0
    # stdoutから直接読み取る (リアルタイム処理)
    if process.stdout:
        for line in process.stdout:
            print(line.strip()) # strip() で改行を除去
            line_count += 1
            if line_count >= 10:
                break # 10行読んだらループを抜ける

    print("\nTerminating the process...")
    process.terminate() # プロセスを終了させる

    # 終了を確認 (少し待つ)
    try:
        process.wait(timeout=1)
        print(f"Process terminated with code: {process.returncode}")
    except subprocess.TimeoutExpired:
        print("Process did not terminate, killing...")
        process.kill() # 強制終了
        process.wait() # killの後、waitで終了を待つのが確実
        print(f"Process killed with code: {process.returncode}")

except FileNotFoundError:
    print("Error: 'yes' command not found.")
except Exception as e:
    print(f"An error occurred: {e}")
      

このように Popen とそのメソッドを組み合わせることで、外部プロセスをより柔軟に、非同期的に制御することが可能になります。

セキュリティに関する注意点 🛡️

subprocess モジュールは非常に強力ですが、使い方を誤ると重大なセキュリティリスクを引き起こす可能性があります。特に注意が必要なのが shell=True オプションの使用です。

シェルインジェクションの危険性 (shell=True)

shell=True を指定すると、渡されたコマンド文字列がシステムのシェルによって解釈・実行されます。これは、シェルの便利な機能（パイプ |、リダイレクト >、コマンド連結 ; や && など）を使いたい場合に一見便利に見えます。

しかし、外部からの入力（ユーザー入力、ファイルの内容、ネットワークからのデータなど）をコマンド文字列に含めて shell=True で実行すると、悪意のあるコードが実行される「シェルインジェクション」脆弱性に繋がる可能性があります。

例えば、ユーザーに入力されたファイル名を削除するつもりのコードを考えてみましょう。

import subprocess

# !!! 危険なコード例 !!!
filename = input("削除するファイル名を入力してください: ")

# ユーザー入力をそのままコマンド文字列に埋め込んでいる
command = f"rm {filename}"
print(f"実行しようとしているコマンド: '{command}'")

# shell=True で実行
try:
    # この形式は絶対に避けるべき！
    subprocess.run(command, shell=True, check=True)
    print(f"ファイル '{filename}' を削除しました。")
except FileNotFoundError:
    print("Error: 'rm' command not found.")
except subprocess.CalledProcessError as e:
    print(f"コマンドの実行に失敗しました: {e}")
except Exception as e:
    print(f"予期せぬエラー: {e}")
      

このコードで、もし悪意のあるユーザーがファイル名として my_file.txt; rm -rf / のような文字列を入力したらどうなるでしょうか？ シェルはこれを2つのコマンドとして解釈します。

1. rm my_file.txt
2. rm -rf /

結果として、意図しないコマンド (rm -rf /) が実行され、システムに壊滅的なダメージを与える可能性があります 😱。

安全な代替策：引数リストの使用 (shell=False)

シェルインジェクションを防ぐための最も重要な対策は、shell=True を避け、コマンドと引数を文字列のリストとして渡すことです。これが subprocess モジュールのデフォルト (shell=False) であり、推奨される方法です。

import subprocess
import shlex # シェルライクな分割に使える

# 安全なコード例
filename = input("削除するファイル名を入力してください: ")

# ユーザー入力は単なる引数として扱われる
command_list = ['rm', filename]
print(f"実行しようとしているコマンドリスト: {command_list}")

try:
    # shell=False (デフォルト) でリストとして渡す
    subprocess.run(command_list, check=True)
    print(f"ファイル '{filename}' を削除しました。")
except FileNotFoundError:
    # 'rm' コマンド自体が見つからないか、指定されたファイル名が不正な場合
    # (例: ';' を含むファイル名は通常ないので、そのような入力は引数として渡され、
    #  'rm' が「そのようなファイルはない」というエラーを出す可能性が高い)
    print(f"Error: 'rm' command not found or invalid argument '{filename}'")
except subprocess.CalledProcessError as e:
    # rm コマンドが失敗した場合 (ファイルが存在しない、権限がないなど)
    print(f"コマンドの実行に失敗しました (Stderr: {e.stderr})")
except Exception as e:
    print(f"予期せぬエラー: {e}")

# もしユーザー入力に空白などが含まれる可能性がある場合、そのままリストに入れるのが安全
# 例: filename = "my file with spaces.txt" -> command_list = ['rm', 'my file with spaces.txt']
# シェルが解釈しないので、ファイル名が正しく 'rm' コマンドに渡される

# どうしても文字列から安全に分割したい場合は shlex.split() を使う
# user_input = "rm 'my file with spaces.txt'"
# command_list = shlex.split(user_input) # -> ['rm', 'my file with spaces.txt']
# subprocess.run(command_list, check=True)
# ただし、ユーザー入力を shlex.split にかけるのは推奨されない。リストで直接組み立てるべき。
      

shell=False の場合、subprocess はシェルを介さずに直接OSのシステムコールを使ってプロセスを生成します。リストの各要素は、OSに対して「これがコマンド名」「これが第一引数」「これが第二引数」というように明確に渡されます。そのため、引数に含まれる特殊文字（;, |, &, > など）はシェルのメタ文字として解釈されず、単なる文字列としてコマンドに渡されます。これにより、シェルインジェクションのリスクを根本的に排除できます 👍。

結論として、外部からの入力をコマンドの一部として使用する場合は、絶対に shell=True を使わず、コマンドと引数をリスト形式で渡してください。 シェルの機能が必要な場合でも、Python側でパイプライン処理を実装するなど、他の方法を検討すべきです。

その他のセキュリティ考慮事項

• コマンドのパス: コマンド名を絶対パスで指定するか、信頼できる PATH 環境変数を設定することで、意図しないコマンドが実行されるリスクを減らせます。
• 実行権限: スクリプトを実行するユーザーの権限が必要最小限であることを確認してください。
• 入出力のサニタイズ: 外部プロセスとの間でやり取りするデータ（特に外部からの入力に基づく場合）は、適切に検証・サニタイズすることが望ましいです。

古い関数について (call, check_call, check_output) 🕰️

subprocess.run() が導入される前（Python 3.5より前）は、以下の関数がよく使われていました。現在でも利用可能ですが、多くの場合 run() を使う方が引数が統一されており、柔軟性が高いため推奨されます。

これらの古い関数は、run() 関数の特定のユースケースを簡潔に書くために存在していました。しかし、run() 関数はこれらの機能をすべてカバーし、さらに多くのオプション（text, input, timeout, stderrのキャプチャなど）を提供します。

特別な理由がない限り、新規のコードでは subprocess.run() を使用し、必要に応じて Popen を利用するのが良いでしょう。

よくあるユースケース 🛠️

subprocess モジュールは様々な場面で活用できます。ここではいくつかの具体的なユースケースを見てみましょう。

1. システム情報の取得

OSが提供するコマンドを使ってシステム情報を取得します。

import subprocess
import platform

def get_system_info():
    info = {}
    system = platform.system()
    info['OS'] = f"{system} {platform.release()}"

    try:
        if system == "Linux":
            # CPU情報 (lscpu)
            result = subprocess.run(['lscpu'], capture_output=True, text=True, check=True)
            # 簡単な抽出 (より正確には正規表現などを使う)
            for line in result.stdout.splitlines():
                if line.startswith("Model name:"):
                    info['CPU'] = line.split(":", 1)[1].strip()
                    break
            # メモリ情報 (free -h)
            result = subprocess.run(['free', '-h'], capture_output=True, text=True, check=True)
            lines = result.stdout.splitlines()
            if len(lines) > 1:
                 # ヘッダ行の次の行 (Mem:) の total, used, free を取得
                 mem_parts = lines[1].split()
                 if len(mem_parts) >= 4:
                     info['Memory (Total/Used/Free)'] = f"{mem_parts[1]} / {mem_parts[2]} / {mem_parts[3]}"

        elif system == "Darwin": # macOS
            # CPU情報 (sysctl)
            result = subprocess.run(['sysctl', '-n', 'machdep.cpu.brand_string'], capture_output=True, text=True, check=True)
            info['CPU'] = result.stdout.strip()
            # メモリ情報 (sysctl) - 物理メモリサイズ
            result = subprocess.run(['sysctl', '-n', 'hw.memsize'], capture_output=True, text=True, check=True)
            mem_bytes = int(result.stdout.strip())
            info['Memory (Total)'] = f"{mem_bytes / (1024**3):.2f} GiB" # GiBに変換

        elif system == "Windows":
            # CPU情報 (wmic)
            result = subprocess.run(['wmic', 'cpu', 'get', 'Name'], capture_output=True, text=True, check=True, encoding='cp932') # Windowsはエンコーディング注意
            lines = result.stdout.splitlines()
            if len(lines) > 1:
                info['CPU'] = lines[1].strip()
            # メモリ情報 (wmic)
            result = subprocess.run(['wmic', 'ComputerSystem', 'get', 'TotalPhysicalMemory'], capture_output=True, text=True, check=True, encoding='cp932')
            lines = result.stdout.splitlines()
            if len(lines) > 1:
                 mem_bytes = int(lines[1].strip())
                 info['Memory (Total)'] = f"{mem_bytes / (1024**3):.2f} GiB"

    except FileNotFoundError as e:
        print(f"Required command not found: {e}")
    except subprocess.CalledProcessError as e:
        print(f"Command failed: {e.cmd}, Stderr: {e.stderr}")
    except Exception as e:
        print(f"An error occurred: {e}")

    return info

if __name__ == "__main__":
    system_info = get_system_info()
    print("--- System Information ---")
    for key, value in system_info.items():
        print(f"{key}: {value}")
    print("-------------------------")
      

※ Windowsでの wmic コマンドは、実行環境やOSバージョンによってエンコーディング (例: cp932) の指定が必要な場合があります。

2. Gitコマンドの実行

バージョン管理システムGitの操作をPythonから行います。

import subprocess
import os

def get_git_branch(repo_path):
    if not os.path.isdir(os.path.join(repo_path, '.git')):
        return None, "Not a git repository"
    try:
        # git branch --show-current (Git 2.22以降)
        # または git rev-parse --abbrev-ref HEAD
        result = subprocess.run(
            ['git', 'branch', '--show-current'],
            cwd=repo_path, # リポジトリのパスで実行
            capture_output=True,
            text=True,
            check=True
        )
        return result.stdout.strip(), None
    except FileNotFoundError:
        return None, "Git command not found"
    except subprocess.CalledProcessError as e:
        # ブランチがない、などのエラー
        return None, f"Git command failed: {e.stderr.strip()}"
    except Exception as e:
         return None, f"An unexpected error: {e}"

def git_pull(repo_path):
    if not os.path.isdir(os.path.join(repo_path, '.git')):
        return False, "Not a git repository"
    try:
        print(f"Running 'git pull' in {repo_path}...")
        result = subprocess.run(
            ['git', 'pull'],
            cwd=repo_path,
            capture_output=True, # 出力を見るためにキャプチャ
            text=True,
            check=True # エラーがあれば例外発生
        )
        print("Git pull successful.")
        print("Output:\n", result.stdout)
        return True, result.stdout
    except FileNotFoundError:
        return False, "Git command not found"
    except subprocess.CalledProcessError as e:
        print(f"Git pull failed:")
        print(f"Stderr: {e.stderr}")
        return False, e.stderr
    except Exception as e:
         print(f"An unexpected error: {e}")
         return False, str(e)

if __name__ == "__main__":
    repo_directory = "." # カレントディレクトリを対象とする場合
    branch, error = get_git_branch(repo_directory)
    if error:
        print(f"Error getting branch: {error}")
    else:
        print(f"Current git branch: {branch}")

    # プルを実行 (注意: 実行するとリポジトリが更新されます)
    # success, message = git_pull(repo_directory)
    # if not success:
    #     print("Failed to pull repository.")
      

3. 外部プログラムの実行と結果利用

別の言語で書かれた計算プログラムを実行し、その結果をPythonで利用します。

例として、標準入力から数値を受け取り、その合計を標準出力に返す簡単な外部プログラム (calculator.py とする) を考えます。

# calculator.py (外部プログラムの例)
import sys

total = 0
for line in sys.stdin:
    try:
        total += float(line.strip())
    except ValueError:
        # 数値に変換できない行は無視 (またはエラー出力)
        pass
print(total)
      

この calculator.py を subprocess で呼び出すPythonスクリプトは以下のようになります。

# main_script.py
import subprocess
import sys

numbers_to_sum = "10\n20.5\n30\n-5.5\n" # \n 区切りで数値を入力

try:
    result = subprocess.run(
        [sys.executable, 'calculator.py'], # Pythonインタプリタ経由で実行
        input=numbers_to_sum,
        capture_output=True,
        text=True,
        check=True # calculator.py がエラー終了したら例外
    )

    # 結果 (標準出力) を数値に変換
    total_sum = float(result.stdout.strip())
    print(f"The sum calculated by external script is: {total_sum}")

except FileNotFoundError:
    print("Error: Python interpreter or calculator.py not found.")
except subprocess.CalledProcessError as e:
    print(f"Calculator script failed:")
    print(f"  Return code: {e.returncode}")
    print(f"  Stderr: {e.stderr}")
except ValueError:
    print(f"Error: Could not convert the script output '{result.stdout.strip()}' to float.")
except Exception as e:
    print(f"An unexpected error occurred: {e}")
      

これらの例は、subprocess がいかに多様なタスクに応用できるかを示しています。ファイル操作、データ処理、他のツールとの連携など、可能性は無限大です ✨。

まとめ 🏁

Pythonの subprocess モジュールは、外部コマンドやプロセスを実行・管理するための強力で柔軟なツールです。この記事では、以下の主要な点について解説しました。

• 基本的なコマンド実行には subprocess.run() を使うのが推奨される。
• コマンドと引数はリスト形式で渡すのが安全 (shell=False)。
• capture_output=True や stdout=subprocess.PIPE, stderr=subprocess.PIPE で出力をキャプチャできる。
• text=True で入出力を文字列として扱える。
• input 引数で標準入力にデータを渡せる。
• check=True でコマンド失敗時に CalledProcessError 例外を発生させられる。
• timeout 引数で実行時間制限を設定できる。
• パイプライン処理や非同期処理には subprocess.Popen クラスが有効。
• shell=True はシェルインジェクションのリスクがあるため、原則として避けるべき。

subprocess モジュールを使いこなすことで、Pythonスクリプトの可能性は大きく広がります。システムの自動化、他のツールとの連携、複雑なワークフローの構築など、様々な場面で役立つはずです。

ただし、特にセキュリティに関しては常に注意を払い、安全な方法で利用することを心がけてください。外部プロセスとのやり取りは、予期せぬ動作や脆弱性の原因となりうるため、慎重な設計とエラーハンドリングが不可欠です。

この記事が、皆さんの subprocess モジュールの理解を深め、活用の一助となれば幸いです。Happy coding! 🎉