Python Subprocess — Running External Commands

The Story That Explains Subprocess

Every operating system runs work as processes. When your Python script needs to run ls -la, invoke ffmpeg to transcode a video, call git status, or execute any other executable on the system, it uses the subprocess module to fork a child process, hand it the command, and communicate with it through three streams: stdin, stdout, and stderr.

The Foundation — What Actually Happens

When you call subprocess.run(["ls", "-la"]), Python asks the operating system to:

subprocess.run — The 90% Case

subprocess.run() is your default. It runs a command, waits for it to finish, and returns a CompletedProcess object with the exit code and (optionally) captured output. This is what you should reach for first.

import subprocess

# Simplest possible call — just run it, wait, done
result = subprocess.run(["echo", "Hello, Mohit!"])
print(f"return code: {result.returncode}")

Capturing Output — capture_output + text

import subprocess

result = subprocess.run(
    ["ls", "-la", "/tmp"],
    capture_output=True,        # capture stdout + stderr
    text=True                    # decode bytes to str
)

print("--- STDOUT ---")
print(result.stdout[:200])
print("--- STDERR ---")
print(result.stderr or "(none)")
print(f"exit code: {result.returncode}")

Handling Errors — check=True and returncode

Every process ends with an exit code: 0 for success, any non-zero value for failure. By default subprocess.run does not raise on non-zero exits — you must check explicitly, or pass check=True.

import subprocess

# Command that will fail — no such file
try:
    result = subprocess.run(
        ["cat", "/does/not/exist"],
        capture_output=True,
        text=True,
        check=True                # raise on non-zero exit
    )
except subprocess.CalledProcessError as e:
    print(f"Command failed with code {e.returncode}")
    print(f"stderr: {e.stderr.strip()}")
    print(f"cmd:    {e.cmd}")

Timeouts — Never Wait Forever

A hung process will freeze your script indefinitely. Always set a timeout when you're not sure how long a command will take.

import subprocess

try:
    result = subprocess.run(
        ["ping", "-c", "100", "google.com"],
        capture_output=True,
        text=True,
        timeout=3                # seconds — kill after this
    )
    print(result.stdout)
except subprocess.TimeoutExpired as e:
    print(f"Timed out after {e.timeout}s")
    print(f"Partial output: {e.stdout[:100] if e.stdout else 'none'}")

subprocess.run — Key Arguments Reference

Passing stdin — sending data to the child

import subprocess

# Sort a list of names by piping stdin
names = "charlie\nalice\nbob\ndave"

result = subprocess.run(
    ["sort"],
    input=names,                # fed to child's stdin
    capture_output=True,
    text=True,
    check=True
)
print(result.stdout)

Custom env & cwd

import subprocess, os

# Run a script in a specific directory with a modified env
custom_env = os.environ.copy()
custom_env["MY_API_KEY"] = "secret-token-123"
custom_env["PYTHONUNBUFFERED"] = "1"

result = subprocess.run(
    ["bash", "-c", "echo $MY_API_KEY; pwd"],
    cwd="/tmp",
    env=custom_env,
    capture_output=True,
    text=True
)
print(result.stdout)

Popen — Low-Level Control for Streaming & Long-Running Jobs

subprocess.run is built on top of subprocess.Popen. Use Popen directly when you need to:

Streaming stdout line by line

import subprocess

# Simulate a long-running task that emits output over time
p = subprocess.Popen(
    ["bash", "-c",
     "for i in 1 2 3 4 5; do echo 'line '$i; sleep 1; done"],
    stdout=subprocess.PIPE,
    stderr=subprocess.STDOUT,       # merge stderr into stdout
    text=True,
    bufsize=1                        # line-buffered
)

# Read one line at a time as it's produced
for line in p.stdout:
    print(f"[live] {line.rstrip()}")

exit_code = p.wait()
print(f"\ndone, exit={exit_code}")

Pipes — Chaining Commands Like the Shell

The shell command ps aux | grep python | wc -l chains three programs: ps's output feeds grep's input; grep's output feeds wc's input. In Python you build the same chain by connecting one process's stdout to another's stdin.

Building the pipeline in Python

import subprocess

# Equivalent of:  ps aux | grep python | wc -l
p1 = subprocess.Popen(
    ["ps", "aux"],
    stdout=subprocess.PIPE
)
p2 = subprocess.Popen(
    ["grep", "python"],
    stdin=p1.stdout,                # feed p1's output
    stdout=subprocess.PIPE
)
p3 = subprocess.Popen(
    ["wc", "-l"],
    stdin=p2.stdout,                # feed p2's output
    stdout=subprocess.PIPE,
    text=True
)

# CRITICAL: close upstream stdouts in the parent so SIGPIPE works
p1.stdout.close()
p2.stdout.close()

count = p3.communicate()[0].strip()
print(f"python processes: {count}")

The shortcut — do it all in one line

import subprocess

# Get the disk usage of /var by grepping df -h output
result = subprocess.run(
    ["grep", "/var"],
    input=subprocess.run(["df", "-h"], capture_output=True, text=True).stdout,
    capture_output=True,
    text=True
)
print(result.stdout)

communicate() — Send and Receive in One Shot

When you spawn with Popen and need to send input and then read all output, use communicate(). It writes the input, closes stdin, reads all of stdout and stderr, and waits for the child to exit — all safely, without deadlocking.

import subprocess

p = subprocess.Popen(
    ["python3"],                # interactive Python
    stdin=subprocess.PIPE,
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    text=True
)

script = """
import math
print('sqrt(2) =', math.sqrt(2))
print('pi     =', math.pi)
"""

stdout, stderr = p.communicate(input=script, timeout=5)
print("--- output ---")
print(stdout)
print(f"exit: {p.returncode}")

The shell=True Trap — Security & Injection

shell=True runs your command through /bin/sh -c "...". Convenient, but if any part of the command string comes from user input, you have a shell injection vulnerability.

Real-World Example — Git Status Dashboard

Let's build a small utility that scans multiple git repositories and reports which have uncommitted changes. Uses subprocess.run, check=False, timeouts, cwd, and error handling — all the pieces you need in production.

import subprocess
from pathlib import Path

def git_status(repo_path: Path) -> dict:
    """Return a summary of the git status of one repo."""
    try:
        # --porcelain gives machine-readable output
        result = subprocess.run(
            ["git", "status", "--porcelain", "-b"],
            cwd=repo_path,
            capture_output=True,
            text=True,
            timeout=10,
            check=True
        )
    except subprocess.CalledProcessError as e:
        return {"name": repo_path.name, "error": e.stderr.strip()}
    except subprocess.TimeoutExpired:
        return {"name": repo_path.name, "error": "timed out"}
    except FileNotFoundError:
        return {"name": repo_path.name, "error": "git not installed"}

    lines = result.stdout.splitlines()
    branch = lines[0].replace("## ", "") if lines else "?"
    changes = [l for l in lines[1:] if l.strip()]

    return {
        "name":    repo_path.name,
        "branch":  branch,
        "dirty":   len(changes) > 0,
        "changes": len(changes)
    }

# Scan multiple repos
repos = [Path("~/work/api").expanduser(),
         Path("~/work/web").expanduser(),
         Path("~/work/docs").expanduser()]

for repo in repos:
    info = git_status(repo)
    if "error" in info:
        print(f"{info['name']:12s}  ✗ {info['error']}")
    else:
        status = "DIRTY" if info["dirty"] else "clean"
        print(f"{info['name']:12s}  {status:6s}  ({info['changes']} changes)  [{info['branch']}]")

Real-World Example — Live Log Tailing with Popen

Here's a small tool that streams the last 100 lines of a log file and follows new entries live — like tail -f — while adding colour-tagged severity levels.

import subprocess

def colorize(line: str) -> str:
    if "ERROR" in line:
        return f"\033[31m{line}\033[0m"       # red
    if "WARN" in line:
        return f"\033[33m{line}\033[0m"       # yellow
    if "INFO" in line:
        return f"\033[36m{line}\033[0m"       # cyan
    return line

def tail_log(path: str, lines: int = 100):
    p = subprocess.Popen(
        ["tail", "-n", str(lines), "-f", path],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True,
        bufsize=1
    )
    try:
        for line in p.stdout:
            print(colorize(line.rstrip()))
    except KeyboardInterrupt:
        print("\nStopping tail…")
        p.terminate()                        # polite SIGTERM first
        try:
            p.wait(timeout=2)
        except subprocess.TimeoutExpired:
            p.kill()                        # SIGKILL if it ignores us
            p.wait()

tail_log("/var/log/app.log")

run vs Popen — Which to Use When

Common Pitfalls

Golden Rules