Stage 7 · Master

Python for Operators

Running Subprocesses

subprocess.run, Popen, capture output, handle errors, and run shell commands.

7 min readPython for DevOpsMaster

subprocess.run — The Default

subprocess.run is the recommended way to run external commands. It is simple, predictable, and handles most use cases. Use it instead of os.system or os.popen, which are deprecated.

Basic subprocess.run usage


import subprocess

# Run a command and check the exit code
result = subprocess.run(["ls", "-la", "/etc/hosts"])
print(result.returncode)  # 0 = success

# Run a command in a specific directory
result = subprocess.run(
    ["git", "status"],
    cwd="/path/to/repo",
)

# Pass environment variables
result = subprocess.run(
    ["make", "build"],
    env={"PATH": "/usr/local/bin:/usr/bin:/bin"},
)

Always pass the command as a list of strings, not a single string. This avoids shell injection vulnerabilities and makes argument handling explicit.

Capturing Output

Capture stdout and stderr


import subprocess

# Capture both stdout and stderr
result = subprocess.run(
    ["kubectl", "get", "pods", "-n", "production"],
    capture_output=True,
    text=True,
)

print("STDOUT:", result.stdout)
print("STDERR:", result.stderr)
print("Exit code:", result.returncode)

# Capture only stdout
result = subprocess.run(
    ["docker", "ps", "--format", "{{.Names}}"],
    capture_output=True,
    text=True,
)
container_names = result.stdout.strip().split("\n")

text=True decodes bytes to strings using the default encoding. Without it, result.stdout and result.stderr are bytes objects.

Buffer limits

capture_output buffers the entire output in memory. If a command produces gigabytes of output, you will run out of memory. For large output, read stdout line by line with Popen.

Popen for Complex Flows

When you need streaming output, piped commands, or long-running processes, use Popen directly.

Stream output line by line


import subprocess

# Stream lines as they are produced
process = subprocess.Popen(
    ["journalctl", "-f", "-u", "nginx"],
    stdout=subprocess.PIPE,
    text=True,
)

for line in process.stdout:
    print(line.rstrip())
    if "error" in line.lower():
        print("ALERT: error detected!")

Popen gives you a file-like object for stdout. Iterating over it reads lines as they arrive, without buffering the entire output.

shell=True vs shell=False

Aspect	shell=False (default)	shell=True
Command format	List: ['ls', '-la']	String: 'ls -la'
Shell expansion	No globbing or pipes	Full shell features
Security	No injection risk	Vulnerable to injection
Performance	Faster (no shell fork)	Slightly slower

When shell=True is necessary


import subprocess

# Shell features: pipes, globbing, environment variables
result = subprocess.run(
    "echo $HOSTNAME && uptime | awk '{print $1}'",
    shell=True,
    capture_output=True,
    text=True,
)

# Safer alternative — avoid shell=True when possible
result = subprocess.run(
    ["sh", "-c", "echo $HOSTNAME && uptime | awk '{print $1}'"],
    capture_output=True,
    text=True,
)

Use shell=True only when you need shell features like pipes and globbing. The safer alternative passes the command through sh -c with a list.

Error Handling

Check return codes and handle failures


import subprocess

def run_command(cmd: list[str], check: bool = True) -> subprocess.CompletedProcess:
    """Run a command with optional error checking."""
    try:
        result = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            check=check,
        )
        return result
    except subprocess.CalledProcessError as e:
        print(f"Command failed: {cmd}")
        print(f"Exit code: {e.returncode}")
        print(f"STDERR: {e.stderr}")
        raise

# Raises CalledProcessError if exit code != 0
result = run_command(["docker", "build", "-t", "myapp", "."])

# Don't raise on failure — handle it yourself
result = run_command(["ping", "-c", "1", "unreachable.host"], check=False)
if result.returncode != 0:
    print("Host unreachable, skipping")

check=True (the default for subprocess.run) raises CalledProcessError on non-zero exit codes. Set check=False to handle errors manually.

Practical Examples

Common DevOps subprocess patterns


import subprocess
from pathlib import Path

# Run a script and capture its output
result = subprocess.run(
    ["python3", "scripts/check_health.py"],
    capture_output=True,
    text=True,
    timeout=30,
)

# Deploy with Terraform
result = subprocess.run(
    ["terraform", "apply", "-auto-approve"],
    cwd=Path("/infra/terraform"),
    env={"TF_VAR_env": "production", **dict(__import__('os').environ)},
    capture_output=True,
    text=True,
)

# Run kubectl and parse JSON output
import json

result = subprocess.run(
    ["kubectl", "get", "pods", "-o", "json", "-n", "default"],
    capture_output=True,
    text=True,
)
pods = json.loads(result.stdout)
for pod in pods["items"]:
    print(pod["metadata"]["name"], pod["status"]["phase"])

timeout raises subprocess.TimeoutExpired if the command takes too long. Always set a timeout for commands that could hang indefinitely.

Use shutil.which to check if a command exists

shutil.which('docker') returns the path to the docker binary or None. Use it to check if a tool is installed before trying to run it.

Mark this lesson complete to store local progress and unlock a cleaner resume path the next time you visit.