Shell Commands and Subprocess

What You'll Learn

How to run shell commands from Python, capture their output, handle errors, and do it safely.

subprocess.run — The Modern Way

subprocess.run() is the recommended approach for running commands:

import subprocess

# Run a command and wait for it to finish
result = subprocess.run(["ls", "-la", "/tmp"], capture_output=True, text=True)

print(result.returncode)   # 0 = success
print(result.stdout)       # command output
print(result.stderr)       # error output

Always use a list of strings — never a shell string with user input:

# ❌ DANGEROUS — command injection possible
user_dir = input("Directory: ")
subprocess.run(f"ls {user_dir}", shell=True)   # user could type: "; rm -rf /"

# ✅ SAFE — arguments are always separate from command
subprocess.run(["ls", user_dir])

Common Options

import subprocess

# Capture stdout and stderr
result = subprocess.run(
    ["git", "log", "--oneline", "-5"],
    capture_output=True,  # = stdout=PIPE, stderr=PIPE
    text=True,            # decode bytes to str automatically
    timeout=30,           # raise TimeoutExpired if too slow
    cwd="/path/to/repo",  # working directory
)

# Check for failure — raises CalledProcessError if non-zero exit
result = subprocess.run(
    ["python3", "-m", "pytest", "tests/"],
    check=True    # raises exception on failure
)

# Pass environment variables
import os
env = {**os.environ, "MY_VAR": "value"}
result = subprocess.run(["./script.sh"], env=env, capture_output=True, text=True)

Handling Output

import subprocess
import sys


def run_command(args: list[str], cwd: str = None) -> tuple[int, str, str]:
    """Run a command and return (exit_code, stdout, stderr)."""
    result = subprocess.run(
        args,
        capture_output=True,
        text=True,
        timeout=60,
        cwd=cwd,
    )
    return result.returncode, result.stdout, result.stderr


exit_code, out, err = run_command(["git", "status"])

if exit_code != 0:
    print(f"Command failed (exit {exit_code}):\n{err}", file=sys.stderr)
else:
    print(out)

Handling Errors

from subprocess import run, CalledProcessError, TimeoutExpired


def safe_run(args: list[str]) -> str | None:
    try:
        result = run(args, capture_output=True, text=True, timeout=30, check=True)
        return result.stdout.strip()

    except CalledProcessError as e:
        print(f"Command failed (exit {e.returncode}):", file=sys.stderr)
        print(e.stderr, file=sys.stderr)
        return None

    except TimeoutExpired:
        print(f"Command timed out after 30s: {args}", file=sys.stderr)
        return None

    except FileNotFoundError:
        print(f"Command not found: {args[0]}", file=sys.stderr)
        return None

Streaming Output (Large Output)

For commands that produce a lot of output, stream line by line:

import subprocess
import sys


def stream_command(args: list[str]) -> int:
    """Run a command and stream its output to stdout in real time."""
    process = subprocess.Popen(
        args,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,   # merge stderr into stdout
        text=True,
        bufsize=1,   # line-buffered
    )

    for line in process.stdout:
        print(line, end="", flush=True)

    process.wait()
    return process.returncode


exit_code = stream_command(["python3", "-m", "pytest", "-v", "tests/"])
sys.exit(exit_code)

Common Shell Operations in Python

Instead of running shell commands, use Python's stdlib where possible:

from pathlib import Path
import shutil
import os

# Instead of: subprocess.run(["cp", src, dst])
shutil.copy2(src, dst)

# Instead of: subprocess.run(["mv", src, dst])
Path(src).rename(dst)

# Instead of: subprocess.run(["rm", "-rf", dir])
shutil.rmtree(dir)

# Instead of: subprocess.run(["mkdir", "-p", path])
Path(path).mkdir(parents=True, exist_ok=True)

# Instead of: subprocess.run(["ls", dir])
list(Path(dir).iterdir())

# Instead of: subprocess.run(["find", dir, "-name", "*.py"])
list(Path(dir).rglob("*.py"))

Useful Command Patterns

import subprocess

# Get git commit hash
result = subprocess.run(["git", "rev-parse", "--short", "HEAD"],
                        capture_output=True, text=True)
commit = result.stdout.strip()

# Check if a command exists
import shutil
if shutil.which("ffmpeg") is None:
    raise RuntimeError("ffmpeg is not installed")

# Run a Python script as a subprocess
result = subprocess.run(
    ["python3", "process.py", "--input", "data.csv"],
    capture_output=True, text=True, check=True
)

# SSH command
result = subprocess.run(
    ["ssh", "user@server", "systemctl", "status", "nginx"],
    capture_output=True, text=True, timeout=10
)

# Check disk space
result = subprocess.run(["df", "-h", "/"], capture_output=True, text=True)
print(result.stdout)

Security: Never Use shell=True with User Input

# ❌ Command injection — never do this
user_file = input("Enter filename: ")
subprocess.run(f"cat {user_file}", shell=True)

# ❌ Also dangerous
subprocess.run("cat " + user_file, shell=True)

# ✅ Safe — argument list, not a string
subprocess.run(["cat", user_file])

Only use shell=True when you need shell features (pipes |, globs *) and the command is fully under your control.

Common Mistakes

Mistake	Fix
shell=True with user input	Use a list of strings instead
No timeout=	Commands can hang forever
No check=True	Silently ignores non-zero exit
Using os.system()	Use subprocess.run() instead
Decoding stdout manually	Use text=True

Quick Reference

import subprocess

# Basic
result = subprocess.run(["cmd", "arg"], capture_output=True, text=True)
result.returncode   # 0 = success
result.stdout       # captured output
result.stderr       # captured errors

# With error handling
result = subprocess.run([...], check=True)   # raises CalledProcessError

# With timeout
result = subprocess.run([...], timeout=30)   # raises TimeoutExpired

# Stream output
proc = subprocess.Popen([...], stdout=subprocess.PIPE, text=True)
for line in proc.stdout:
    print(line, end="")
proc.wait()

# Check if program exists
import shutil
if shutil.which("ffmpeg"):
    ...

What's Next

→ Lesson 5: Scheduling and Exit Codes