Python - subprocess Module

The Python subprocess module allows you to spawn new system processes, connect to their input/output/error pipes, and obtain their return codes. It replaces legacy functions like os.system(), os.popen(), and commands.

1. Modern Standard: subprocess.run()

For most synchronous command executions in Python 3.7+, use subprocess.run().

Basic Execution & Capturing Output

Pass commands as a list of strings. Use capture_output=True and text=True (to decode bytes to str automatically):

import subprocess

result = subprocess.run(
    ["ls", "-lh", "/var/log"],
    capture_output=True,
    text=True,
    check=True  # Raises CalledProcessError if returncode != 0
)

print(f"Exit Code: {result.returncode}")
print(f"Stdout:\n{result.stdout}")

2. Common Patterns & Options

Error Handling (check=True)

try:
    result = subprocess.run(
        ["git", "checkout", "non-existent-branch"],
        capture_output=True,
        text=True,
        check=True
    )
except subprocess.CalledProcessError as e:
    print(f"Command failed with return code {e.returncode}")
    print(f"Error message: {e.stderr}")

Passing Input to stdin

result = subprocess.run(
    ["grep", "ERROR"],
    input="INFO: All ok\nERROR: Database down\nDEBUG: Step 1",
    capture_output=True,
    text=True
)
print(result.stdout) # "ERROR: Database down\n"

Timeouts & Custom Environment

import os

custom_env = {**os.environ, "CUSTOM_VAR": "value"}

try:
    result = subprocess.run(
        ["sleep", "10"],
        timeout=3, # Raise TimeoutExpired after 3 seconds
        cwd="/tmp",
        env=custom_env
    )
except subprocess.TimeoutExpired:
    print("Process timed out!")

Security Warning: Avoid shell=True

# Unsafe: Subject to Shell Injection vulnerabilities!
user_input = "file.txt; rm -rf /"
subprocess.run(f"cat {user_input}", shell=True) # NEVER DO THIS

# Safe: Arguments passed as a list bypass shell expansion
subprocess.run(["cat", user_input])

3. Advanced Streaming & Pipelines: subprocess.Popen

When you need non-blocking asynchronous execution, streaming real-time logs, or Unix-style piping (cmd1 | cmd2), use Popen.

Real-Time Streaming Output Line-by-Line

process = subprocess.Popen(
    ["ping", "-c", "4", "google.com"],
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    text=True
)

# Stream stdout as it arrives
for line in process.stdout:
    print(f"[LOG] {line.strip()}")

process.wait() # Wait for process to complete

Piping Processes Together (cat file.txt | grep error | wc -l)

p1 = subprocess.Popen(["cat", "access.log"], stdout=subprocess.PIPE)
p2 = subprocess.Popen(["grep", "404"], stdin=p1.stdout, stdout=subprocess.PIPE)
p3 = subprocess.Popen(["wc", "-l"], stdin=p2.stdout, stdout=subprocess.PIPE, text=True)

# Allow p1 and p2 to receive SIGPIPE if downstream process exits
p1.stdout.close()
p2.stdout.close()

output, _ = p3.communicate()
print(f"Total 404s: {output.strip()}")

Summary Cheatsheet

Task	Recommended Function	Key Arguments
Run command & get return code	subprocess.run(cmd)	check=False
Capture string output	subprocess.run(cmd)	capture_output=True, text=True
Raise error on non-zero exit	subprocess.run(cmd)	check=True
Prevent hanging commands	subprocess.run(cmd)	timeout=seconds
Real-time stream processing	subprocess.Popen(cmd)	stdout=subprocess.PIPE, text=True
Pipe between processes	subprocess.Popen(cmd)	stdin=prev_proc.stdout