Python subprocessの使い方｜run・Popenの違いと例外処理の実例

subprocessとは、Pythonから別のプログラム（ls、git、別のPythonスクリプトなど）を起動し、その出力や終了コードを受け取るための標準モジュールです。 ふだんは subprocess.run() を使い、実行中の出力を逐次読みたい・複数のコマンドをつなぎたいときに subprocess.Popen を使う、と覚えると迷いません。

この記事では、関数の使い分け表、run() の引数一覧、例外（CalledProcessError / TimeoutExpired / FileNotFoundError）の扱い、Popen でのリアルタイム読み取りとパイプ、communicate() を使う理由、shell=True を安全に扱う書き方、Windowsとの違い、asyncio での並行実行までを扱います。コードはすべて Linux（Ubuntu 24.04）上の Python 3.11.15 で実行し、一部は Python 3.14.7 でも確認した結果を載せています。

subprocessの早見表

関数の使い分け

run() の引数一覧

subprocess.run() の基本：コマンドを実行して結果を受け取る

capture_output=True, text=True を付けると、出力を文字列として受け取れます。 戻り値は CompletedProcess です。

import subprocess

r = subprocess.run(["ls", "-l", "/etc/hostname"], capture_output=True, text=True)
print(r)
print("returncode:", r.returncode)
print("stdout:", r.stdout, end="")
print("stderr:", repr(r.stderr))

CompletedProcess(args=['ls', '-l', '/etc/hostname'], returncode=0, stdout='-rw-r--r-- 1 root root 3 Oct  2 00:08 /etc/hostname\n', stderr='')
returncode: 0
stdout: -rw-r--r-- 1 root root 3 Oct  2 00:08 /etc/hostname
stderr: ''

text=True を付けないとbytesになる

r = subprocess.run(["echo", "バイト"], capture_output=True)
print(r.stdout)
r = subprocess.run(["echo", "日本語"], capture_output=True, encoding="utf-8")
print(r.stdout, end="")

b'\xe3\x83\x90\xe3\x82\xa4\xe3\x83\x88\n'
日本語

text=True は環境の既定の文字コードで変換し、encoding="utf-8" は文字コードを固定します。 WindowsやOSの設定が異なる環境でも同じ結果にしたいときは encoding を指定すると確実です。

文字列をそのまま渡すと FileNotFoundError

shell=False（既定）で "ls -l files" のような1つの文字列を渡すと、その文字列全体が1つのコマンド名として扱われます。

try:
    subprocess.run("ls -l files")
except FileNotFoundError as e:
    print("FileNotFoundError:", e)

FileNotFoundError: [Errno 2] No such file or directory: 'ls -l files'

["ls", "-l", "files"] のようにリストにするか、手元の文字列を分けたいなら shlex.split() を使います。

import shlex
print(shlex.split("grep -n 'hello world' memo.txt"))

['grep', '-n', 'hello world', 'memo.txt']

終了コードと標準エラーの取り方

コマンドが失敗しても、check=True を付けない限り例外にはなりません。 returncode と stderr で判定します。

r = subprocess.run(["ls", "/nothing"], capture_output=True, text=True)
print("returncode:", r.returncode)
print("stderr:", r.stderr, end="")

returncode: 2
stderr: ls: cannot access '/nothing': No such file or directory

標準出力と標準エラーを分ける・まとめる

r = subprocess.run(["sh", "-c", "echo out; echo err >&2; exit 3"],
                   capture_output=True, text=True)
print(r.returncode, repr(r.stdout), repr(r.stderr))

r = subprocess.run(["sh", "-c", "echo out; echo err >&2"],
                   stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True)
print(repr(r.stdout))

3 'out\n' 'err\n'
'out\nerr\n'

stderr=subprocess.STDOUT で標準エラーを標準出力にまとめられます。 ログに時系列で残したいときに便利です。なお capture_output=True と stdout= / stderr= は同時に指定できません。

入力を渡す・場所と環境変数を変える

import os

r = subprocess.run(["sort"], input="banana\napple\ncherry\n", capture_output=True, text=True)
print(r.stdout, end="")

r = subprocess.run(["pwd"], cwd="/tmp", capture_output=True, text=True)
print(r.stdout, end="")

r = subprocess.run(["sh", "-c", "echo $APP_MODE"],
                   env={**os.environ, "APP_MODE": "test"}, capture_output=True, text=True)
print(r.stdout, end="")

r = subprocess.run(["sh", "-c", "echo $HOME"],
                   env={"APP_MODE": "test"}, capture_output=True, text=True)
print(repr(r.stdout))

apple
banana
cherry
/tmp
test
'\n'

最後の例のように、env に新しい辞書だけを渡すと HOME や PATH などがすべて消えます。 追加したいだけなら {**os.environ, "キー": "値"} と元の環境変数を広げてから足しましょう。

例外処理：CalledProcessError・TimeoutExpired・FileNotFoundError

subprocessで押さえる例外は主に3つです。 共通の親クラスは subprocess.SubprocessError（FileNotFoundError は OSError の仲間）です。

CalledProcessError

try:
    subprocess.run(["ls", "/nothing"], capture_output=True, text=True, check=True)
except subprocess.CalledProcessError as e:
    print("CalledProcessError:", e)
    print("e.returncode:", e.returncode)
    print("e.stderr:", e.stderr, end="")

CalledProcessError: Command '['ls', '/nothing']' returned non-zero exit status 2.
e.returncode: 2
e.stderr: ls: cannot access '/nothing': No such file or directory

capture_output=True を付けておくと、例外の中に stderr が入るので、原因をそのままログに出せます。

シグナルで止まった場合は終了コードが負の数になり、メッセージも変わります。

r = subprocess.run(["sh", "-c", "kill -TERM $$"])
print(r.returncode)
try:
    subprocess.run(["sh", "-c", "kill -TERM $$"], check=True)
except subprocess.CalledProcessError as e:
    print(e)

-15
Command '['sh', '-c', 'kill -TERM $$']' died with <Signals.SIGTERM: 15>.

TimeoutExpired

run() の timeout を超えると、子プロセスを止めてから TimeoutExpired を送出します。

try:
    subprocess.run(["sleep", "5"], timeout=1)
except subprocess.TimeoutExpired as e:
    print("TimeoutExpired:", e)

Python 3.11.15 と 3.14.7 で表示が異なりました。

# Python 3.11.15
TimeoutExpired: Command '['sleep', '5']' timed out after 0.9999899499999856 seconds
# Python 3.14.7
TimeoutExpired: Command '['sleep', '5']' timed out after 1 seconds

メッセージの文字列で判定せず、例外の型と e.timeout 属性で判定するとバージョンに左右されません。

FileNotFoundError

try:
    subprocess.run(["no-such-cmd"])
except FileNotFoundError as e:
    print("FileNotFoundError:", e)

FileNotFoundError: [Errno 2] No such file or directory: 'no-such-cmd'

まとめて扱うコピペ用テンプレート

import logging
import subprocess

logger = logging.getLogger(__name__)

def run_cmd(args: list[str], timeout: float = 30) -> str:
    """コマンドを実行し、標準出力を返す。失敗時は原因をログに出して再送出する。"""
    try:
        r = subprocess.run(args, capture_output=True, text=True,
                           check=True, timeout=timeout)
        return r.stdout
    except FileNotFoundError:
        logger.error("コマンドが見つかりません: %s", args[0])
        raise
    except subprocess.TimeoutExpired as e:
        logger.error("%s 秒でタイムアウト: %s", e.timeout, args)
        raise
    except subprocess.CalledProcessError as e:
        logger.error("終了コード %s: %s\n%s", e.returncode, args, e.stderr)
        raise

例外処理そのものの書き方（raise from など）は、Pythonの例外処理の記事で詳しく扱っています。ここではsubprocess固有の3つを押さえておけば十分です。

古い関数との対応

call() check_call() check_output() は今も使えますが、run() で同じことが書けます。

Popenで出力を1行ずつリアルタイムに読む

run() は終わるまで結果を返しません。時間のかかる処理の進み具合を逐次表示したいときは Popen で stdout を1行ずつ読みます。

import subprocess, sys, time

script = "import time\nfor i in range(3):\n    print(f'step {i}', flush=True)\n    time.sleep(0.5)\n"
t0 = time.monotonic()
with subprocess.Popen([sys.executable, "-c", script],
                      stdout=subprocess.PIPE, text=True, bufsize=1) as p:
    for line in p.stdout:
        print(f"[{time.monotonic()-t0:.1f}s] {line}", end="")
print("returncode:", p.returncode)

[0.1s] step 0
[0.6s] step 1
[1.1s] step 2
returncode: 0

with を使うと、ブロックを抜けるときにパイプを閉じてプロセスの終了を待ってくれるので、returncode が確定します。

まとめて届くときは子プロセス側のバッファ

出力が最後にまとめて届く場合、原因は子プロセスが出力をためている（バッファリング）ことです。 Pythonの子プロセスは、出力先がパイプだとまとめて書き出します。flush なしのスクリプトで比べました（環境変数 PYTHONUNBUFFERED を外した状態）。

# python -c script（flushなし）
[1.5s] step 0
[1.5s] step 1
[1.5s] step 2
# python -u -c script
[0.0s] step 0
[0.5s] step 1
[1.0s] step 2

Pythonなら -u・PYTHONUNBUFFERED=1・print(..., flush=True)、多くのLinuxコマンドなら stdbuf -oL コマンド（GNU coreutils）で逐次出力にできます。 出力を捨てたいときは stdout=subprocess.DEVNULL を指定します。

Popen の主なメソッド

Popenでコマンドをパイプでつなぐ

シェルの A | B は、A の stdout を B の stdin に渡すことで再現できます。 シェルを使わないので、値の扱いが安全です。

p1 = subprocess.Popen(["printf", "c\\nb\\na\\nb\\n"], stdout=subprocess.PIPE)
p2 = subprocess.Popen(["sort", "-u"], stdin=p1.stdout, stdout=subprocess.PIPE, text=True)
p1.stdout.close()   # p2 が先に終わったとき p1 に通知が届くようにする
out, _ = p2.communicate()
print(out, end="")
print(p1.wait(), p2.returncode)

a
b
c
0 0

p1.stdout.close() は公式ドキュメントでも推奨されている一行で、後ろのコマンドが先に終わったときに前のコマンドが SIGPIPE を受け取って止まれるようにします。

communicate() でデッドロックを避ける

stdout=PIPE を指定したまま wait() で待つと、出力が多い場合に子プロセスが書き込み待ちで止まり、親も終了待ちで止まります。 パイプには一定の容量（Linuxでは一般に64KiB）しかないためです。20万バイトを出力する子プロセスで確かめました。

import subprocess, sys

big = "import sys\nsys.stdout.write('x'*200000)\n"

# 直す前：wait() で待つ
p = subprocess.Popen([sys.executable, "-c", big], stdout=subprocess.PIPE)
try:
    p.wait(timeout=2)
except subprocess.TimeoutExpired as e:
    print("wait():", e)
    p.kill(); p.wait(); p.stdout.close()

# 直した後：communicate() で読みながら待つ
p = subprocess.Popen([sys.executable, "-c", big], stdout=subprocess.PIPE)
out, err = p.communicate(timeout=5)
print("communicate():", len(out), "bytes, returncode", p.returncode)

wait(): Command '['/usr/bin/python3', '-c', "import sys\nsys.stdout.write('x'*200000)\n"]' timed out after 2 seconds
communicate(): 200000 bytes, returncode 0

wait() は2秒待っても終わらず、communicate() はすぐ200,000バイトを受け取って終了しました。 stdout と stderr の両方を PIPE にする場合も、communicate() が両方を同時に読むので詰まりません。

communicate() のタイムアウトの書き方

communicate(timeout=...) でタイムアウトしても子プロセスは止まりません。 公式ドキュメントにあるとおり、kill() してからもう一度 communicate() を呼んで後始末します。

p = subprocess.Popen(["sleep", "10"])
try:
    p.communicate(timeout=1)
except subprocess.TimeoutExpired:
    p.kill()
    out, err = p.communicate()
    print("killed, returncode", p.returncode)

killed, returncode -9

shell=True を安全に扱う：リスト渡しとshlex.quote

shell=True はコマンド文字列をシェル（Linuxでは /bin/sh）に解釈させます。外部から受け取った値をそのまま文字列に埋め込むと、; などの記号が「次のコマンド」として解釈されます（コマンドインジェクション）。 防ぐ側の確認として、作業用フォルダで再現しました。

files/
├── memo 2026.txt
└── report.txt

直す前：f文字列で組み立てて shell=True

name = "report.txt; echo INJECTED > files/pwned.txt"   # 利用者から受け取った値の想定
r = subprocess.run(f"ls files/{name}", shell=True, capture_output=True, text=True)
print("shell=True:", repr(r.stdout), r.returncode)
print(subprocess.run("ls files", shell=True, capture_output=True, text=True).stdout, end="")

shell=True: 'files/report.txt\n' 0
memo 2026.txt
pwned.txt
report.txt

ファイル名のつもりで渡した値の後半が別のコマンドとして実行され、pwned.txt が作られました。

直した後1：リストで渡す（推奨）

r = subprocess.run(["ls", f"files/{name}"], capture_output=True, text=True)
print("list:", r.returncode, r.stderr, end="")

list: 2 ls: cannot access 'files/report.txt; echo INJECTED > files/pwned.txt': No such file or directory

リストで渡すとシェルを通らないので、値全体が1つの引数（ファイル名）として扱われます。 空白を含むファイル名もそのまま渡せます。

r = subprocess.run(["ls", "files/memo 2026.txt"], capture_output=True, text=True)
print(r.stdout, end="")
r = subprocess.run("ls files/memo 2026.txt", shell=True, capture_output=True, text=True)
print(r.returncode, r.stderr, end="")

files/memo 2026.txt
2 ls: cannot access 'files/memo': No such file or directory
ls: cannot access '2026.txt': No such file or directory

直した後2：どうしてもシェルが必要なら shlex.quote

パイプやリダイレクトをシェルに任せたいときは、埋め込む値を shlex.quote() で囲みます。

import shlex
q = shlex.quote(f"files/{name}")
print("quoted:", q)
r = subprocess.run(f"ls {q}", shell=True, capture_output=True, text=True)
print("shlex.quote:", r.returncode, r.stderr, end="")
print(shlex.join(["ls", "files/memo 2026.txt"]))

quoted: 'files/report.txt; echo INJECTED > files/pwned.txt'
shlex.quote: 2 ls: cannot access 'files/report.txt; echo INJECTED > files/pwned.txt': No such file or directory
ls 'files/memo 2026.txt'

オプションとして解釈されないように -- を付ける

リスト渡しでも、値が - で始まるとコマンドのオプションとして読まれます。多くのコマンドは -- 以降をオプションとして扱わないので、外部の値の前に置いておくと確実です。

r = subprocess.run(["ls", "--", "-la"], capture_output=True, text=True)
print(r.returncode, r.stderr, end="")

2 ls: cannot access '-la': No such file or directory

Windowsとの違い（要点）

以下は公式ドキュメントに沿った要点で、この記事ではWindows上での実行はしていません。

asyncio.create_subprocess_exec の最小例

asyncioを使っているコードでは、asyncio.create_subprocess_exec() を使うと、待ち時間の間にほかの処理を進められます。 1秒かかるコマンドを3つ同時に実行しました。

import asyncio, time

async def run(*cmd):
    proc = await asyncio.create_subprocess_exec(
        *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE)
    out, err = await proc.communicate()
    return cmd[-1], proc.returncode, out.decode().strip()

async def main():
    t0 = time.monotonic()
    results = await asyncio.gather(
        run("sh", "-c", "sleep 1; echo A"),
        run("sh", "-c", "sleep 1; echo B"),
        run("sh", "-c", "sleep 1; echo C"),
    )
    for r in results:
        print(r)
    print(f"elapsed: {time.monotonic() - t0:.1f}s")

asyncio.run(main())

('sleep 1; echo A', 0, 'A')
('sleep 1; echo B', 0, 'B')
('sleep 1; echo C', 0, 'C')
elapsed: 1.0s

3つで合計1.0秒でした。create_subprocess_exec() はリスト渡しに相当し、シェルを通す版は create_subprocess_shell() です。asyncioの詳しい使い方（TaskGroup やタイムアウト）、スレッド・プロセスを使った並列処理は、それぞれの記事で扱っています。

よくある質問

Q. subprocess.run と Popen はどちらを使えばよいですか？ A. まず run() を使い、実行中の出力を逐次読みたい・パイプでつなぎたい・裏で動かし続けたいときだけ Popen を使うのがおすすめです。run() は内部で Popen と communicate() を使っているので、デッドロックの心配もありません。

Q. コマンドが失敗しても例外になりません。 A. check=True を付けると、終了コードが0以外のとき CalledProcessError になります。 付けない場合は r.returncode で判定します。

Q. 出力が None になります。 A. capture_output=True（または stdout=subprocess.PIPE）を付けていないことが原因です。付けない場合、出力はそのまま画面（親プロセスの標準出力）に流れます。

Q. shell=True はいつ使ってよいですか？ A. コマンド文字列がすべてプログラム内で固定され、外部の値を含まない場合に限るとわかりやすくなります。値を含める場合はリスト渡しにするか、shlex.quote() で囲みます。

まとめ

• 基本は subprocess.run(リスト, capture_output=True, text=True)。戻り値は CompletedProcess
• 失敗を例外にするなら check=True、止まらない処理には timeout。例外は CalledProcessError / TimeoutExpired / FileNotFoundError の3つを押さえる
• 終了コードは 0が成功、負の数はシグナルで停止（-15 はSIGTERM）
• 逐次出力は Popen で1行ずつ読む。まとめて届くなら子プロセスのバッファ（python -u、stdbuf -oL）
• PIPE を使うなら wait() ではなく communicate()。20万バイトの出力で違いを確認
• 引数はリストで渡し、shell=True に外部の値を混ぜない。必要なら shlex.quote() と --
• Windowsでは encoding="utf-8" の指定と、組み込みコマンドの shell=True を意識する

株式会社RJCでは、Pythonで外部コマンドを扱う技術を仲間と学び合っています。こうした知識を共有しながら、日々技術を磨いています。

株式会社RJC ― SI事業・SES事業・AI駆動開発。RJCは一緒に成長を楽しめる会社です。