Skip to content

Python subprocess：Pythonから外部コマンドを実行する（完全ガイド）

Soren Atelier
Name
Soren Atelier

Updated on

Pythonスクリプトでは、外部プログラムを呼び出す必要がよくあります。ファイルを圧縮するためにシェルコマンドを実行したり、バージョン管理のためにgitを呼び出したり、ffmpegのようなシステムユーティリティで動画を処理したり、データパイプラインの一部としてコンパイル済みバイナリを実行したりすることがあります。しかし、os.system()やバッククォート風のハックに頼ると、壊れやすく、危険で、問題が起きたときにデバッグ不能なコードになってしまいます。

苦痛はすぐに増幅します。出力は取得できずに闇に消え、戻り値コードが無視されることでエラーは静かに見過ごされます。スペースやセミコロンを含むユーザー入力のファイル名が1つあるだけで、無害なスクリプトがシェルインジェクション脆弱性に変わり得ます。さらに、subprocessがハングすると、Pythonプログラム全体が一緒に止まります—タイムアウトも、復旧も、説明もありません。

Pythonのsubprocessモジュールは、その標準的な解決策です。os.system()、os.popen()、そして非推奨となったcommandsモジュールのような古い関数を、プロセス起動、出力キャプチャ、エラー処理、タイムアウト設定、パイプライン構築までを一貫したAPIで提供する仕組みに置き換えます。このガイドでは、これを効果的かつ安全に使いこなすために必要なことをすべて扱います。

📚

    subprocess.run()でクイックスタート

    subprocess.run()関数はPython 3.5で導入され、外部コマンドを実行する推奨手段です。コマンドを実行して終了まで待ち、CompletedProcessオブジェクトを返します。

    import subprocess
 
# Run a simple command
result = subprocess.run(["ls", "-la"], capture_output=True, text=True)
 
print(result.stdout)       # standard output as a string
print(result.stderr)       # standard error as a string
print(result.returncode)   # 0 means success

    主要なパラメータ:

    • capture_output=True はstdoutとstderrをキャプチャします（stdout=subprocess.PIPE, stderr=subprocess.PIPEと同等）
    • text=True は出力をbytesではなく文字列としてデコードします
    • コマンドは文字列のリストとして渡し、各引数は別々の要素にします
    import subprocess
 
# Run a command with arguments
result = subprocess.run(
    ["python", "--version"],
    capture_output=True,
    text=True
)
print(result.stdout.strip())  # e.g., "Python 3.12.1"

    コマンドの「リスト」指定と「文字列」指定を理解する

    subprocess.run()の第1引数はリストにも文字列にもできます。この違いは正しさとセキュリティの両面で重要です。

    リスト形式（推奨）

    リストの各要素が個別の引数になります。Pythonはシェル解釈を介さず、OSへ直接引数を渡します。

    import subprocess
 
# Each argument is a separate list element
result = subprocess.run(
    ["grep", "-r", "TODO", "/home/user/project"],
    capture_output=True,
    text=True
)
print(result.stdout)

    スペース、クォート、特殊文字を含むファイル名でも、各引数がそのまま渡されるため正しく動作します。

    import subprocess
 
# Filename with spaces -- works correctly as a list element
result = subprocess.run(
    ["cat", "my file with spaces.txt"],
    capture_output=True,
    text=True
)

    文字列形式（shell=Trueが必要）

    1つの文字列を渡す場合、shell=Trueが必要になります。これにより、システムシェル（Unixでは/bin/sh、Windowsではcmd.exe）がコマンドを解釈します。

    import subprocess
 
# String form requires shell=True
result = subprocess.run(
    "ls -la | grep '.py'",
    shell=True,
    capture_output=True,
    text=True
)
print(result.stdout)

    これによりパイプ（|）、リダイレクト（>）、グロブ（*.py）、環境変数展開（$HOME）といったシェル機能を使えます。しかし同時に重大なセキュリティリスクも導入します。これは後述のセキュリティ節で扱います。

    出力をキャプチャする

    stdoutとstderrを別々にキャプチャ

    import subprocess
 
result = subprocess.run(
    ["python", "-c", "import sys; print('out'); print('err', file=sys.stderr)"],
    capture_output=True,
    text=True
)
 
print(f"stdout: {result.stdout}")   # "out\n"
print(f"stderr: {result.stderr}")   # "err\n"

    stderrをstdoutにまとめる

    すべての出力を1つのストリームにまとめたい場合があります。そのときはstderr=subprocess.STDOUTを使います。

    import subprocess
 
result = subprocess.run(
    ["python", "-c", "import sys; print('out'); print('err', file=sys.stderr)"],
    stdout=subprocess.PIPE,
    stderr=subprocess.STDOUT,
    text=True
)
 
print(result.stdout)  # Contains both "out\n" and "err\n"

    出力を破棄する

    出力を抑制するにはsubprocess.DEVNULLに送ります。

    import subprocess
 
# Run silently -- discard all output
result = subprocess.run(
    ["apt-get", "update"],
    stdout=subprocess.DEVNULL,
    stderr=subprocess.DEVNULL
)

    バイナリ出力

    text=Trueを付けないと、生のbytesとして受け取れます。画像や圧縮ファイルのようなバイナリデータに有用です。

    import subprocess
 
# Capture binary output (e.g., from curl)
result = subprocess.run(
    ["curl", "-s", "https://example.com/image.png"],
    capture_output=True
)
 
image_bytes = result.stdout  # bytes object
print(f"Downloaded {len(image_bytes)} bytes")

    エラー処理

    戻り値コードを手動でチェックする

    デフォルトでは、subprocess.run()はコマンド失敗時に例外を投げません。自分でreturncodeを確認する必要があります。

    import subprocess
 
result = subprocess.run(
    ["ls", "/nonexistent/path"],
    capture_output=True,
    text=True
)
 
if result.returncode != 0:
    print(f"Command failed with code {result.returncode}")
    print(f"Error: {result.stderr}")

    check=Trueで失敗時に自動で例外

    check=Trueを指定すると、戻り値が0以外のときにsubprocess.CalledProcessErrorが送出されます。

    import subprocess
 
try:
    result = subprocess.run(
        ["ls", "/nonexistent/path"],
        capture_output=True,
        text=True,
        check=True
    )
except subprocess.CalledProcessError as e:
    print(f"Command failed with return code {e.returncode}")
    print(f"stderr: {e.stderr}")
    print(f"stdout: {e.stdout}")

    これは「必ず成功してほしい」コマンドに対する推奨パターンです。失敗を黙って無視せず、明示的に扱うことを強制できます。

    コマンドが見つからない場合の扱い

    実行ファイルが存在しない場合、PythonはFileNotFoundErrorを送出します。

    import subprocess
 
try:
    result = subprocess.run(
        ["nonexistent_command"],
        capture_output=True,
        text=True
    )
except FileNotFoundError:
    print("Command not found -- is it installed and in PATH?")

    タイムアウト

    長時間実行やハングするプロセスは、スクリプトを永遠にブロックし得ます。timeout（秒）を指定すると、時間内に終了しなかった場合にプロセスを停止し、subprocess.TimeoutExpiredを送出します。

    import subprocess
 
try:
    result = subprocess.run(
        ["sleep", "30"],
        timeout=5,
        capture_output=True,
        text=True
    )
except subprocess.TimeoutExpired:
    print("Process timed out after 5 seconds")

    ネットワークコマンド、外部API、ハングし得る処理では必須です。

    import subprocess
 
def run_with_timeout(cmd, timeout_seconds=30):
    """Run a command with timeout and error handling."""
    try:
        result = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            timeout=timeout_seconds,
            check=True
        )
        return result.stdout
    except subprocess.TimeoutExpired:
        print(f"Command timed out after {timeout_seconds}s: {' '.join(cmd)}")
        return None
    except subprocess.CalledProcessError as e:
        print(f"Command failed (code {e.returncode}): {e.stderr}")
        return None
    except FileNotFoundError:
        print(f"Command not found: {cmd[0]}")
        return None
 
# Usage
output = run_with_timeout(["ping", "-c", "4", "example.com"], timeout_seconds=10)
if output:
    print(output)

    プロセスに入力を渡す

    inputパラメータを使うと、プロセスのstdinへデータを送れます。

    import subprocess
 
# Send text to stdin
result = subprocess.run(
    ["grep", "error"],
    input="line 1\nerror on line 2\nline 3\nerror on line 4\n",
    capture_output=True,
    text=True
)
 
print(result.stdout)
# "error on line 2\nerror on line 4\n"

    これは、シェルコマンドでデータをパイプする典型パターンの代替になります。

    import subprocess
import json
 
# Send JSON to a processing command
data = {"name": "Alice", "score": 95}
json_string = json.dumps(data)
 
result = subprocess.run(
    ["python", "-c", "import sys, json; d = json.load(sys.stdin); print(d['name'])"],
    input=json_string,
    capture_output=True,
    text=True
)
 
print(result.stdout.strip())  # "Alice"

    環境変数

    デフォルトでは、subprocessは現在の環境を引き継ぎます。必要に応じて変更できます。

    import subprocess
import os
 
# Add or override environment variables
custom_env = os.environ.copy()
custom_env["API_KEY"] = "secret123"
custom_env["DEBUG"] = "true"
 
result = subprocess.run(
    ["python", "-c", "import os; print(os.environ.get('API_KEY'))"],
    env=custom_env,
    capture_output=True,
    text=True
)
 
print(result.stdout.strip())  # "secret123"

    常にos.environ.copy()をベースにしてください。既存の環境を含まないdictを渡すと、PATHやHOMEなどの継承変数が消えてしまい、コマンドが壊れる原因になります。

    作業ディレクトリ

    cwdパラメータはsubprocessの作業ディレクトリを設定します。

    import subprocess
 
# Run git status in a specific repository
result = subprocess.run(
    ["git", "status", "--short"],
    cwd="/home/user/my-project",
    capture_output=True,
    text=True
)
 
print(result.stdout)

    subprocess.run() vs Popen：どちらを使うべきか

    subprocess.run()はsubprocess.Popenの便利ラッパーです。ほとんどの用途ではrun()で十分です。次が必要な場合にPopenを使います。

    • 出力のリアルタイムストリーミング（生成されるそばから1行ずつ）
    • 実行中プロセスとの対話（入力を送る、ループで出力を読む）
    • 複数プロセスをつなぐマルチステップパイプライン構築
    • ブロッキングせずに実行し、プロセスのライフサイクルを手動制御する

    比較表

    Featuresubprocess.run()subprocess.Popenos.system()
    RecommendedYes (Python 3.5+)Yes (advanced)No (deprecated pattern)
    Capture outputYes (capture_output=True)Yes (via PIPE)No
    Return valueCompletedProcess objectPopen process objectExit code (int)
    Timeout supportYes (timeout param)Manual (via wait/communicate)No
    Error checkingcheck=True raises exceptionManualMust parse exit code
    Input to stdininput parametercommunicate() or stdin.write()No
    Real-time outputNo (waits for completion)Yes (stream line by line)Output goes to terminal
    PipelinesLimited (single command)Yes (chain multiple Popen)Yes (via shell string)
    SecuritySafe with list argsSafe with list argsShell injection risk
    Shell featuresOnly with shell=TrueOnly with shell=TrueAlways uses shell

    高度編：subprocess.Popen

    Popenはプロセスのライフサイクルを完全に制御できます。コンストラクタは即座にプロセスを開始し、対話可能なPopenオブジェクトを返します。

    基本的なPopenの使い方

    import subprocess
 
proc = subprocess.Popen(
    ["ls", "-la"],
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    text=True
)
 
stdout, stderr = proc.communicate()  # Wait for completion and get output
print(f"Return code: {proc.returncode}")
print(stdout)

    出力をリアルタイムにストリームする

    run()とは異なり、Popenなら出力を生成されるそばから1行ずつ読めます。

    import subprocess
 
proc = subprocess.Popen(
    ["ping", "-c", "5", "example.com"],
    stdout=subprocess.PIPE,
    text=True
)
 
# Read output line by line as it arrives
for line in proc.stdout:
    print(f"[LIVE] {line.strip()}")
 
proc.wait()  # Wait for process to finish
print(f"Exit code: {proc.returncode}")

    進捗表示やリアルタイムログが欲しい長時間コマンドでは特に重要です。

    import subprocess
import sys
 
def run_with_live_output(cmd):
    """Run a command and stream its output in real time."""
    proc = subprocess.Popen(
        cmd,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        text=True,
        bufsize=1  # Line-buffered
    )
 
    output_lines = []
    for line in proc.stdout:
        line = line.rstrip()
        print(line)
        output_lines.append(line)
 
    proc.wait()
    return proc.returncode, "\n".join(output_lines)
 
# Usage
code, output = run_with_live_output(["pip", "install", "requests"])
print(f"\nFinished with exit code: {code}")

    パイプラインを構築する

    複数コマンドを接続し、あるプロセスのstdoutを別プロセスのstdinへ渡します。

    import subprocess
 
# Equivalent to: cat /var/log/syslog | grep "error" | wc -l
p1 = subprocess.Popen(
    ["cat", "/var/log/syslog"],
    stdout=subprocess.PIPE
)
 
p2 = subprocess.Popen(
    ["grep", "error"],
    stdin=p1.stdout,
    stdout=subprocess.PIPE
)
 
# Allow p1 to receive SIGPIPE if p2 exits early
p1.stdout.close()
 
p3 = subprocess.Popen(
    ["wc", "-l"],
    stdin=p2.stdout,
    stdout=subprocess.PIPE,
    text=True
)
 
p2.stdout.close()
 
output, _ = p3.communicate()
print(f"Error count: {output.strip()}")

    p1.stdout.close()は重要です。p2が早期終了した場合にp1がSIGPIPEを受け取れるようになり、デッドロック防止につながります。

    対話的なプロセス通信

    import subprocess
 
# Start a Python REPL as a subprocess
proc = subprocess.Popen(
    ["python", "-i"],
    stdin=subprocess.PIPE,
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    text=True
)
 
# Send commands and get results
stdout, stderr = proc.communicate(input="print(2 + 2)\nprint('hello')\n")
print(f"stdout: {stdout}")
print(f"stderr: {stderr}")

    shell=True：強力だが危険

    shell=Trueを設定するとコマンドがシステムシェル経由で実行され、シェル機能を有効化できますが、セキュリティリスクも導入します。

    shell=Trueが便利な場面

    import subprocess
 
# Shell features: pipes, redirects, globbing, env vars
result = subprocess.run(
    "ls *.py | wc -l",
    shell=True,
    capture_output=True,
    text=True
)
print(f"Python files: {result.stdout.strip()}")
 
# Environment variable expansion
result = subprocess.run(
    "echo $HOME",
    shell=True,
    capture_output=True,
    text=True
)
print(result.stdout.strip())

    シェルインジェクション問題

    未サニタイズのユーザー入力をshell=Trueへ渡してはいけません。

    import subprocess
 
# DANGEROUS -- shell injection vulnerability
user_input = "file.txt; rm -rf /"  # malicious input
subprocess.run(f"cat {user_input}", shell=True)  # Executes "rm -rf /"!
 
# SAFE -- use list form without shell=True
subprocess.run(["cat", user_input])  # Treats entire string as filename

    動的な値を含めたままshell=Trueを使う必要があるなら、shlex.quote()を使います。

    import subprocess
import shlex
 
user_input = "file with spaces.txt; rm -rf /"
safe_input = shlex.quote(user_input)
 
# shlex.quote wraps in single quotes, neutralizing shell metacharacters
result = subprocess.run(
    f"cat {safe_input}",
    shell=True,
    capture_output=True,
    text=True
)

    ただし最も安全なのは、shell=Trueを避け、シェル機能をPythonで置き換えることです。

    import subprocess
import glob
 
# Instead of: subprocess.run("ls *.py | wc -l", shell=True)
py_files = glob.glob("*.py")
print(f"Python files: {len(py_files)}")
 
# Instead of: subprocess.run("cat file1.txt file2.txt > combined.txt", shell=True)
with open("combined.txt", "w") as outfile:
    result = subprocess.run(
        ["cat", "file1.txt", "file2.txt"],
        stdout=outfile
    )

    セキュリティのベストプラクティス

    PracticeDoDon't
    Command format["cmd", "arg1", "arg2"]f"cmd {user_input}" with shell=True
    User inputshlex.quote() if shell is neededString concatenation into commands
    Shell modeshell=False (default)shell=True with untrusted input
    Executable pathFull paths like /usr/bin/gitRely on PATH for security-critical code
    Input validationValidate and sanitize before passingPass raw user input to commands
    import subprocess
import shlex
from pathlib import Path
 
def safe_file_operation(filename):
    """Safely run a command with user-supplied filename."""
    # Validate input
    path = Path(filename)
    if not path.exists():
        raise FileNotFoundError(f"File not found: {filename}")
 
    # Check for path traversal
    resolved = path.resolve()
    allowed_dir = Path("/home/user/uploads").resolve()
    if not str(resolved).startswith(str(allowed_dir)):
        raise PermissionError("Access denied: file outside allowed directory")
 
    # Use list form -- no shell injection possible
    result = subprocess.run(
        ["wc", "-l", str(resolved)],
        capture_output=True,
        text=True,
        check=True
    )
    return result.stdout.strip()

    実務的な例

    gitコマンドを実行する

    import subprocess
 
def git_status(repo_path):
    """Get git status for a repository."""
    result = subprocess.run(
        ["git", "status", "--porcelain"],
        cwd=repo_path,
        capture_output=True,
        text=True,
        check=True
    )
    return result.stdout.strip()
 
def git_log(repo_path, n=5):
    """Get last n commit messages."""
    result = subprocess.run(
        ["git", "log", f"--oneline", f"-{n}"],
        cwd=repo_path,
        capture_output=True,
        text=True,
        check=True
    )
    return result.stdout.strip()
 
status = git_status("/home/user/my-project")
if status:
    print("Uncommitted changes:")
    print(status)
else:
    print("Working directory clean")

    ファイルを圧縮・解凍する

    import subprocess
 
def compress_directory(source_dir, output_file):
    """Create a tar.gz archive of a directory."""
    subprocess.run(
        ["tar", "-czf", output_file, "-C", source_dir, "."],
        check=True
    )
    print(f"Created archive: {output_file}")
 
def extract_archive(archive_file, dest_dir):
    """Extract a tar.gz archive."""
    subprocess.run(
        ["tar", "-xzf", archive_file, "-C", dest_dir],
        check=True
    )
    print(f"Extracted to: {dest_dir}")
 
compress_directory("/home/user/data", "/tmp/data_backup.tar.gz")

    システム情報を確認する

    import subprocess
 
def get_disk_usage():
    """Get disk usage summary."""
    result = subprocess.run(
        ["df", "-h", "/"],
        capture_output=True,
        text=True,
        check=True
    )
    return result.stdout
 
def get_memory_info():
    """Get memory usage on Linux."""
    result = subprocess.run(
        ["free", "-h"],
        capture_output=True,
        text=True,
        check=True
    )
    return result.stdout
 
def get_process_list(filter_name=None):
    """List running processes, optionally filtered."""
    cmd = ["ps", "aux"]
    result = subprocess.run(cmd, capture_output=True, text=True, check=True)
 
    if filter_name:
        lines = result.stdout.strip().split("\n")
        header = lines[0]
        matching = [line for line in lines[1:] if filter_name in line]
        return header + "\n" + "\n".join(matching)
 
    return result.stdout
 
print(get_disk_usage())

    外部ツールでデータファイルを処理する

    import subprocess
import csv
import io
 
def sort_csv_by_column(input_file, column_index=1):
    """Sort a CSV file using the system sort command (fast for large files)."""
    result = subprocess.run(
        ["sort", "-t,", f"-k{column_index}", input_file],
        capture_output=True,
        text=True,
        check=True
    )
    return result.stdout
 
def count_lines(filepath):
    """Count lines in a file using wc (faster than Python for huge files)."""
    result = subprocess.run(
        ["wc", "-l", filepath],
        capture_output=True,
        text=True,
        check=True
    )
    return int(result.stdout.strip().split()[0])
 
def search_in_files(directory, pattern, file_type="*.py"):
    """Search for a pattern in files using grep."""
    result = subprocess.run(
        ["grep", "-rn", "--include", file_type, pattern, directory],
        capture_output=True,
        text=True
    )
    # grep returns exit code 1 if no matches found (not an error)
    if result.returncode == 0:
        return result.stdout
    elif result.returncode == 1:
        return ""  # No matches
    else:
        raise subprocess.CalledProcessError(result.returncode, result.args)
 
matches = search_in_files("/home/user/project", "TODO")
if matches:
    print(matches)
else:
    print("No TODOs found")

    自動デプロイスクリプト

    import subprocess
import sys
 
def deploy(repo_path, branch="main"):
    """Simple deployment script using subprocess."""
    steps = [
        (["git", "fetch", "origin"], "Fetching latest changes"),
        (["git", "checkout", branch], f"Switching to {branch}"),
        (["git", "pull", "origin", branch], "Pulling latest code"),
        (["pip", "install", "-r", "requirements.txt"], "Installing dependencies"),
        (["python", "manage.py", "migrate"], "Running migrations"),
        (["python", "manage.py", "collectstatic", "--noinput"], "Collecting static files"),
    ]
 
    for cmd, description in steps:
        print(f"\n--- {description} ---")
        try:
            result = subprocess.run(
                cmd,
                cwd=repo_path,
                capture_output=True,
                text=True,
                check=True,
                timeout=120
            )
            if result.stdout:
                print(result.stdout)
        except subprocess.CalledProcessError as e:
            print(f"FAILED: {e.stderr}")
            sys.exit(1)
        except subprocess.TimeoutExpired:
            print(f"TIMEOUT: {description} took too long")
            sys.exit(1)
 
    print("\nDeployment complete")

    クロスプラットフォームでの注意点

    コマンドはWindowsとUnixで挙動が異なります。移植性のあるコードを書きましょう。

    import subprocess
import platform
 
def run_command(cmd_unix, cmd_windows=None):
    """Run a command with platform awareness."""
    if platform.system() == "Windows":
        cmd = cmd_windows or cmd_unix
        # Windows often needs shell=True for built-in commands
        return subprocess.run(cmd, shell=True, capture_output=True, text=True)
    else:
        return subprocess.run(cmd, capture_output=True, text=True)
 
# List directory contents
result = run_command(
    cmd_unix=["ls", "-la"],
    cmd_windows="dir"
)
print(result.stdout)

    主なプラットフォーム差分:

    FeatureUnix/macOSWindows
    Shell/bin/shcmd.exe
    Path separator/\\
    Built-in commands (dir, copy)Not availableRequire shell=True
    Executable extensionNot needed.exe required sometimes
    Signal handlingFull POSIX signalsLimited
    shlex.quote()WorksUse subprocess.list2cmdline()

    Jupyter Notebooksでsubprocessを実行する

    Jupyter notebooksからシェルコマンドを実行するのは、データサイエンティストにとって一般的なワークフローです。Jupyterには手軽な!command構文がありますが、subprocessを使えば、Pythonコードの中で適切なエラー処理と出力キャプチャができます。

    ノートブック内でsubprocess呼び出しをデバッグする際—特にコマンドが静かに失敗したり、想定外の出力を出したりするとき—RunCell (opens in a new tab)が役立つことがあります。RunCellはノートブックの文脈を理解するJupyter向けAIエージェントです。subprocessコマンドが失敗する理由を診断し、正しい引数を提案し、プラットフォーム固有の癖にも対応できます。ターミナルとノートブックを行き来してシェルコマンドをデバッグする代わりに、RunCellはセル内で直接問題を追跡します。

    import subprocess
 
# In a Jupyter notebook: capture and display command output
result = subprocess.run(
    ["pip", "list", "--format=columns"],
    capture_output=True,
    text=True
)
 
# Display as formatted output in the notebook
print(result.stdout)

    よくある間違いと修正方法

    間違い1：出力キャプチャを忘れる

    import subprocess
 
# Output goes to terminal, not captured
result = subprocess.run(["ls", "-la"])
print(result.stdout)  # None!
 
# Fix: add capture_output=True
result = subprocess.run(["ls", "-la"], capture_output=True, text=True)
print(result.stdout)  # Actual output

    間違い2：shell=Trueなしで文字列を使う

    import subprocess
 
# Fails: string passed without shell=True
# subprocess.run("ls -la")  # FileNotFoundError: "ls -la" is not a program
 
# Fix option 1: use a list
subprocess.run(["ls", "-la"])
 
# Fix option 2: use shell=True (less safe)
subprocess.run("ls -la", shell=True)

    間違い3：エラーを無視する

    import subprocess
 
# Bad: silently continues on failure
result = subprocess.run(["rm", "/important/file"], capture_output=True, text=True)
# ... continues even if rm failed
 
# Good: check=True raises exception on failure
try:
    result = subprocess.run(
        ["rm", "/important/file"],
        capture_output=True,
        text=True,
        check=True
    )
except subprocess.CalledProcessError as e:
    print(f"Failed to delete: {e.stderr}")

    間違い4：Popenでのデッドロック

    import subprocess
 
# DEADLOCK: stdout buffer fills up, process blocks, .wait() waits forever
proc = subprocess.Popen(["command_with_lots_of_output"], stdout=subprocess.PIPE)
proc.wait()  # Deadlock!
 
# Fix: use communicate() which handles buffering
proc = subprocess.Popen(["command_with_lots_of_output"], stdout=subprocess.PIPE, text=True)
stdout, stderr = proc.communicate()  # Safe

    間違い5：エンコーディングを扱わない

    import subprocess
 
# Bytes output can cause issues
result = subprocess.run(["cat", "data.txt"], capture_output=True)
# result.stdout is bytes, not str
 
# Fix: use text=True or encoding parameter
result = subprocess.run(["cat", "data.txt"], capture_output=True, text=True)
 
# For specific encodings:
result = subprocess.run(
    ["cat", "data.txt"],
    capture_output=True,
    encoding="utf-8",
    errors="replace"  # Handle invalid bytes
)

    subprocess.run() パラメータ完全リファレンス

    import subprocess
 
result = subprocess.run(
    args,                    # Command as list or string
    stdin=None,              # Input source (PIPE, DEVNULL, file object, or None)
    stdout=None,             # Output destination
    stderr=None,             # Error destination
    capture_output=False,    # Shorthand for stdout=PIPE, stderr=PIPE
    text=False,              # Decode output as strings (alias: universal_newlines)
    shell=False,             # Run through system shell
    cwd=None,                # Working directory
    timeout=None,            # Seconds before TimeoutExpired
    check=False,             # Raise CalledProcessError on non-zero exit
    env=None,                # Environment variables dict
    encoding=None,           # Output encoding (alternative to text=True)
    errors=None,             # Encoding error handling ('strict', 'replace', 'ignore')
    input=None,              # String/bytes to send to stdin
)

    FAQ

    Pythonのsubprocessモジュールとは何ですか？

    subprocessモジュールは、Pythonスクリプト内から外部コマンドやプログラムを実行するための標準ライブラリです。os.system()、os.popen()、commandsモジュールのような古い手法を置き換えました。新しいプロセスを生成し、入出力/エラーパイプへ接続し、戻り値コードの取得やタイムアウト処理を行うための機能を提供します。基本となるインターフェースは、単純なコマンド実行向けのsubprocess.run()と、リアルタイムI/Oやプロセスパイプラインが必要な高度な用途向けのsubprocess.Popenです。

    subprocess.run()とsubprocess.Popenの違いは何ですか？

    subprocess.run()は高レベルの便利関数で、コマンドを実行して終了まで待ち、出力を含むCompletedProcessオブジェクトを返します。ほとんどのタスクではこれが適切です。一方subprocess.Popenは低レベルのクラスで、プロセスを直接制御できます。出力を1行ずつストリームしたり、対話的に入力を送ったり、複数プロセスのパイプラインを構築したり、プロセスのライフサイクルを手動管理できます。リアルタイム出力が必要な場合や複数プロセス接続が必要な場合はPopenを使います。

    subprocessでshell=Trueは危険ですか？

    はい。信頼できない入力と一緒にshell=Trueを使うと、シェルインジェクション脆弱性になります。shell=Trueを設定すると、コマンド文字列はシステムシェルに渡されて解釈されるため、;、|、&&、$()のようなメタ文字が実行されます。攻撃者が任意コマンドを注入できる可能性があります。安全なデフォルトは、shell=Falseのままコマンドをリストとして渡すことです。どうしてもshell=Trueが必要なら、shlex.quote()で入力をサニタイズし、生のユーザー入力を決して渡さないでください。

    subprocessコマンドの出力をキャプチャするにはどうすればいいですか？

    subprocess.run()にcapture_output=Trueとtext=Trueを指定します。出力はresult.stdout（文字列）に、エラーはresult.stderrに入ります。例: result = subprocess.run(["ls", "-la"], capture_output=True, text=True) のあと result.stdout にアクセスします。text=Trueを付けない場合、出力はbytesで返ります。

    Pythonでsubprocessのタイムアウトを扱うには？

    subprocess.run()にtimeout（秒）を渡します。プロセスがタイムアウトを超えるとPythonが停止させ、subprocess.TimeoutExpiredを送出します。例: subprocess.run(["slow_command"], timeout=30)。Popenの場合はproc.communicate(timeout=30)またはproc.wait(timeout=30)を使います。タイムアウトがあり得るコードは常にtry/exceptで囲ってください。

    subprocessが推奨なのに、なぜos.system()はまだ動くのですか？

    os.system()は形式上は非推奨ではありませんが、レガシーなインターフェースと見なされています。シェル経由でコマンドを実行（shell=True相当）し、出力をキャプチャできず、タイムアウト機構もなく、終了ステータスしか返しません。subprocess.run()はos.system()ができることに加えて、出力キャプチャ、エラー処理、タイムアウト、安全な引数渡しを提供します。新規コードはすべてsubprocessを使うべきです。

    結論

    subprocessモジュールは、外部コマンドを実行するためのPythonの決定版ツールです。単純なコマンド実行にはsubprocess.run()を使いましょう。出力キャプチャ、エラーチェック、タイムアウト、入力の受け渡しを、1つの関数呼び出しで扱えます。subprocess.Popenに手を伸ばすのは、リアルタイム出力ストリーミング、対話的なプロセス通信、またはマルチステップのパイプラインが必要な場合だけで十分です。

    最も重要な習慣は、ユーザー入力を伴ってshell=Trueを使わないことです。コマンドをリストで渡せば、シェルインジェクションのリスクを完全に排除できます。check=Trueで失敗を早期に検知し、timeoutでハングを防ぎ、text=Trueで生bytesではなく文字列として扱いましょう。

    gitの自動化からデータパイプラインのオーケストレーションまで、subprocessはos.system()では得られない制御性と安全性を提供します。これらのパターンを身につければ、どんな外部ツールでも自信を持ってPythonワークフローに統合できます。

    📚