When I first needed to run shell commands from Python, I reached for os.system(). Then a senior engineer sat me down and explained why that was a terrible idea. This is everything I learned about doing it right with subprocess.
Why subprocess?
The subprocess module is Python's modern way to spawn processes and interact with them. It replaces older functions like os.system(), os.spawn*(), and os.popen*() with a unified, safer API.
subprocess.run() Basics
The simplest way to run a command:
import subprocess
# Run a command and wait for it to complete
result = subprocess.run(["ls", "-la"])
print(f"Return code: {result.returncode}")Always pass commands as a list of strings, not a single string. This is safer and more explicit:
# Good: list of arguments
subprocess.run(["git", "status", "--short"])
# Also works, but we'll see why this is dangerous
subprocess.run("git status --short", shell=True)Capturing Output
By default, output goes to your terminal. To capture it:
import subprocess
result = subprocess.run(
["python", "--version"],
capture_output=True, # Capture stdout and stderr
text=True # Decode bytes to string
)
print(f"stdout: {result.stdout}")
print(f"stderr: {result.stderr}")
print(f"return code: {result.returncode}")The capture_output=True is shorthand for stdout=subprocess.PIPE, stderr=subprocess.PIPE.
Getting Just stdout
result = subprocess.run(
["echo", "hello"],
capture_output=True,
text=True
)
output = result.stdout.strip()
print(output) # "hello"Combining stdout and stderr
result = subprocess.run(
["some_command"],
capture_output=True,
stderr=subprocess.STDOUT, # Redirect stderr to stdout
text=True
)
# Both streams in result.stdoutThe shell=True Danger Zone
I learned this the hard way: shell=True is a security footgun.
What shell=True Does
With shell=True, your command runs through the system shell (like /bin/sh). This enables shell features:
# Shell features work
subprocess.run("echo $HOME", shell=True) # Variable expansion
subprocess.run("ls *.py", shell=True) # Glob expansion
subprocess.run("cat file.txt | grep error", shell=True) # PipesThe Security Problem
# DANGER: Command injection!
user_input = "file.txt; rm -rf /"
subprocess.run(f"cat {user_input}", shell=True)
# This runs: cat file.txt; rm -rf /Any user input + shell=True = potential disaster.
Safe Alternative
import shlex
# Use shlex.split() to parse command strings safely
cmd = "git log --oneline -n 5"
subprocess.run(shlex.split(cmd))
# Or just use lists (preferred)
subprocess.run(["git", "log", "--oneline", "-n", "5"])When shell=True is Okay
- You control the entire command string (no user input)
- You actually need shell features (pipes, redirection)
- Scripts where security isn't a concern
Even then, prefer Python's alternatives:
# Instead of shell pipes, use Python
from pathlib import Path
# Shell: cat file.txt | grep error | wc -l
content = Path("file.txt").read_text()
error_lines = [line for line in content.splitlines() if "error" in line]
print(len(error_lines))Popen: Advanced Process Control
subprocess.run() is convenient but waits for the command to finish. For more control, use Popen:
import subprocess
# Start a process without waiting
proc = subprocess.Popen(
["long_running_command"],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True
)
# Do other work...
print("Process started, doing other things...")
# Later, wait for it
stdout, stderr = proc.communicate()
print(f"Process finished with code {proc.returncode}")Popen for Background Processes
import subprocess
import time
# Start a server in the background
server = subprocess.Popen(
["python", "-m", "http.server", "8000"],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL
)
print(f"Server started with PID {server.pid}")
# ... do your work ...
time.sleep(5)
# Clean up
server.terminate()
server.wait()
print("Server stopped")Polling vs Waiting
import subprocess
import time
proc = subprocess.Popen(["sleep", "5"])
while True:
status = proc.poll()
if status is not None:
print(f"Process exited with code {status}")
break
print("Still running...")
time.sleep(1)stdin/stdout/stderr Handling
Writing to stdin
import subprocess
# Send data to a process
result = subprocess.run(
["python", "-c", "import sys; print(sys.stdin.read().upper())"],
input="hello world",
capture_output=True,
text=True
)
print(result.stdout) # "HELLO WORLD\n"With Popen for Interactive Processes
import subprocess
proc = subprocess.Popen(
["cat"],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
text=True
)
# Write and read (small data only!)
stdout, _ = proc.communicate(input="Hello from Python!\n")
print(stdout)Streaming Large Output
For commands that produce lots of output, don't use capture_output (memory hog). Stream instead:
import subprocess
proc = subprocess.Popen(
["find", "/", "-name", "*.py"],
stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL,
text=True
)
for line in proc.stdout:
print(line.strip())
# Process each line without loading everything into memory
proc.wait()Redirecting to Files
import subprocess
with open("output.log", "w") as f:
subprocess.run(["some_command"], stdout=f, stderr=subprocess.STDOUT)Timeouts
Never trust external commands to finish in reasonable time:
import subprocess
try:
result = subprocess.run(
["sleep", "100"],
timeout=5 # Kill after 5 seconds
)
except subprocess.TimeoutExpired:
print("Command took too long!")Timeout with Popen
import subprocess
proc = subprocess.Popen(["slow_command"])
try:
proc.wait(timeout=10)
except subprocess.TimeoutExpired:
proc.kill() # Force kill
proc.wait() # Clean up zombie process
print("Had to kill the process")Error Handling
Check Return Codes
import subprocess
result = subprocess.run(["false"]) # Returns exit code 1
if result.returncode != 0:
print(f"Command failed with code {result.returncode}")check=True for Automatic Exceptions
import subprocess
try:
subprocess.run(
["ls", "nonexistent_file"],
check=True, # Raise exception on non-zero exit
capture_output=True,
text=True
)
except subprocess.CalledProcessError as e:
print(f"Command failed: {e}")
print(f"stderr: {e.stderr}")FileNotFoundError
import subprocess
try:
subprocess.run(["nonexistent_command"])
except FileNotFoundError:
print("Command not found!")Comprehensive Error Handling
import subprocess
def run_command(cmd, timeout=30):
"""Run a command with proper error handling."""
try:
result = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=timeout,
check=True
)
return result.stdout
except FileNotFoundError:
raise RuntimeError(f"Command not found: {cmd[0]}")
except subprocess.TimeoutExpired:
raise RuntimeError(f"Command timed out after {timeout}s")
except subprocess.CalledProcessError as e:
raise RuntimeError(f"Command failed: {e.stderr}")Security Considerations
Never Trust User Input
# BAD: User controls the filename
def process_file(user_filename):
subprocess.run(f"cat {user_filename}", shell=True)
# GOOD: Use list form, validate input
def process_file(user_filename):
# Validate the filename
if not user_filename.isalnum():
raise ValueError("Invalid filename")
subprocess.run(["cat", user_filename])Sanitize Environment Variables
import subprocess
import os
# Pass only what you need, not the entire environment
safe_env = {
"PATH": "/usr/bin:/bin",
"HOME": os.environ.get("HOME", "/tmp")
}
subprocess.run(["some_command"], env=safe_env)Use Full Paths
import subprocess
# More secure: use absolute paths
subprocess.run(["/usr/bin/git", "status"])
# Or verify the command location
import shutil
git_path = shutil.which("git")
if git_path:
subprocess.run([git_path, "status"])Limit Process Capabilities
import subprocess
# Run with restricted resources (Unix)
subprocess.run(
["untrusted_program"],
cwd="/tmp", # Limit working directory
env={"PATH": "/usr/bin"}, # Minimal environment
timeout=10 # Always timeout
)Common Patterns
Chaining Commands (Piping)
Instead of shell pipes, use Python:
import subprocess
# Shell: ps aux | grep python
ps = subprocess.run(
["ps", "aux"],
capture_output=True,
text=True
)
for line in ps.stdout.splitlines():
if "python" in line:
print(line)For true piping (when you need it):
import subprocess
# Connect two processes
ps = subprocess.Popen(
["ps", "aux"],
stdout=subprocess.PIPE
)
grep = subprocess.Popen(
["grep", "python"],
stdin=ps.stdout,
stdout=subprocess.PIPE,
text=True
)
ps.stdout.close() # Allow ps to receive SIGPIPE
output = grep.communicate()[0]
print(output)Running Commands in a Different Directory
import subprocess
result = subprocess.run(
["git", "status"],
cwd="/path/to/repo", # Run in this directory
capture_output=True,
text=True
)Running with Different User (Unix)
import subprocess
import pwd
import os
def demote(user_uid, user_gid):
def result():
os.setgid(user_gid)
os.setuid(user_uid)
return result
# Run as 'nobody' user
pw_record = pwd.getpwnam("nobody")
subprocess.run(
["id"],
preexec_fn=demote(pw_record.pw_uid, pw_record.pw_gid)
)Async Subprocess (Python 3.5+)
import asyncio
async def run_async():
proc = await asyncio.create_subprocess_exec(
"ls", "-la",
stdout=asyncio.subprocess.PIPE
)
stdout, _ = await proc.communicate()
return stdout.decode()
# In an async context
output = asyncio.run(run_async())Quick Reference
| Task | Method |
|---|---|
| Simple command | subprocess.run(["cmd", "arg"]) |
| Capture output | subprocess.run(..., capture_output=True, text=True) |
| Check success | subprocess.run(..., check=True) |
| Set timeout | subprocess.run(..., timeout=30) |
| Send input | subprocess.run(..., input="data") |
| Change directory | subprocess.run(..., cwd="/path") |
| Background process | subprocess.Popen(...) |
| Stream output | for line in Popen(...).stdout: |
Summary
The subprocess module is powerful but requires care:
- Use list arguments instead of string +
shell=True - Never pass unsanitized user input to any command
- Always set timeouts for external commands
- Handle errors explicitly with try/except
- Capture output only when you need it
- Use Popen when you need background processes or streaming
It took me a few scary moments with shell=True to learn these lessons. Hopefully this guide saves you from the same mistakes!