subprocess lets you run external commands from Python. Here's how to use it safely and effectively.

Basic Usage

import subprocess
 
# Simple command
result = subprocess.run(["ls", "-la"])
 
# Capture output
result = subprocess.run(
    ["git", "status"],
    capture_output=True,
    text=True
)
print(result.stdout)

subprocess.run()

The recommended way to run commands:

result = subprocess.run(
    ["command", "arg1", "arg2"],
    capture_output=True,  # Capture stdout/stderr
    text=True,            # Return strings, not bytes
    check=True,           # Raise on non-zero exit
    timeout=30,           # Timeout in seconds
    cwd="/path/to/dir",   # Working directory
    env={"VAR": "value"}, # Environment variables
)
 
# Result attributes
result.returncode  # Exit code
result.stdout      # Captured stdout
result.stderr      # Captured stderr

Capturing Output

# Capture both stdout and stderr
result = subprocess.run(
    ["python", "--version"],
    capture_output=True,
    text=True
)
print(f"stdout: {result.stdout}")
print(f"stderr: {result.stderr}")
 
# Redirect stderr to stdout
result = subprocess.run(
    ["command"],
    stdout=subprocess.PIPE,
    stderr=subprocess.STDOUT,
    text=True
)

Error Handling

import subprocess
 
try:
    result = subprocess.run(
        ["git", "status"],
        check=True,  # Raises CalledProcessError on failure
        capture_output=True,
        text=True
    )
except subprocess.CalledProcessError as e:
    print(f"Command failed with code {e.returncode}")
    print(f"stderr: {e.stderr}")
except subprocess.TimeoutExpired:
    print("Command timed out")
except FileNotFoundError:
    print("Command not found")

Input to Commands

# Pass input via stdin
result = subprocess.run(
    ["grep", "pattern"],
    input="line1\npattern here\nline3",
    capture_output=True,
    text=True
)
 
# From file
with open("input.txt") as f:
    result = subprocess.run(
        ["sort"],
        stdin=f,
        capture_output=True,
        text=True
    )

Shell Commands

# Using shell (be careful with user input!)
result = subprocess.run(
    "ls -la | grep .py",
    shell=True,
    capture_output=True,
    text=True
)
 
# Safer: pipe without shell
ls = subprocess.run(["ls", "-la"], capture_output=True, text=True)
grep = subprocess.run(
    ["grep", ".py"],
    input=ls.stdout,
    capture_output=True,
    text=True
)

⚠️ Warning: Never use shell=True with user-provided input—it's a security risk.

Popen for Complex Cases

For streaming output or interactive processes:

import subprocess
 
# Stream output line by line
process = subprocess.Popen(
    ["tail", "-f", "logfile.log"],
    stdout=subprocess.PIPE,
    text=True
)
 
for line in process.stdout:
    print(f"Log: {line.strip()}")
    if "ERROR" in line:
        process.terminate()
        break
 
# Interactive process
process = subprocess.Popen(
    ["python"],
    stdin=subprocess.PIPE,
    stdout=subprocess.PIPE,
    text=True
)
stdout, stderr = process.communicate("print('Hello')\n")

Pipes Between Processes

# cat file | grep pattern | wc -l
p1 = subprocess.Popen(
    ["cat", "file.txt"],
    stdout=subprocess.PIPE
)
p2 = subprocess.Popen(
    ["grep", "pattern"],
    stdin=p1.stdout,
    stdout=subprocess.PIPE
)
p3 = subprocess.Popen(
    ["wc", "-l"],
    stdin=p2.stdout,
    stdout=subprocess.PIPE,
    text=True
)
 
p1.stdout.close()  # Allow p1 to receive SIGPIPE
p2.stdout.close()
 
output = p3.communicate()[0]

Environment Variables

import os
 
# Inherit and add
env = os.environ.copy()
env["MY_VAR"] = "value"
 
result = subprocess.run(
    ["printenv", "MY_VAR"],
    env=env,
    capture_output=True,
    text=True
)
 
# Replace entirely
result = subprocess.run(
    ["command"],
    env={"PATH": "/usr/bin", "HOME": "/tmp"}
)

Common Patterns

Check if command exists

import shutil
 
def command_exists(cmd):
    return shutil.which(cmd) is not None
 
if command_exists("git"):
    subprocess.run(["git", "status"])

Run with timeout

try:
    result = subprocess.run(
        ["slow_command"],
        timeout=10,
        capture_output=True
    )
except subprocess.TimeoutExpired:
    print("Command took too long")

Get command output as string

def run_command(cmd):
    result = subprocess.run(
        cmd,
        capture_output=True,
        text=True,
        check=True
    )
    return result.stdout.strip()
 
version = run_command(["python", "--version"])

Run silently

subprocess.run(
    ["command"],
    stdout=subprocess.DEVNULL,
    stderr=subprocess.DEVNULL
)

Build command safely

import shlex
 
# Parse string to list (for trusted input)
cmd = shlex.split("git commit -m 'My message'")
# ['git', 'commit', '-m', 'My message']
 
# Quote for shell (if you must use shell=True)
filename = "file with spaces.txt"
safe_name = shlex.quote(filename)

Quick Reference

import subprocess
 
# Simple run
subprocess.run(["cmd", "arg"])
 
# Capture output
result = subprocess.run(
    ["cmd"],
    capture_output=True,
    text=True,
    check=True
)
output = result.stdout
 
# With input
subprocess.run(["cmd"], input="data", text=True)
 
# With timeout
subprocess.run(["cmd"], timeout=30)
 
# Check exit code
if result.returncode != 0:
    print("Failed")
 
# Streaming (Popen)
proc = subprocess.Popen(["cmd"], stdout=subprocess.PIPE)
for line in proc.stdout:
    process(line)

Use subprocess.run() for most cases. Use Popen only when you need streaming or complex process control.

React to this post: