Run shell commands, capture output, handle exit codes, and orchestrate complex automation workflows by executing external tools from Python—the bridge between Python scripts and system commands.
Sometimes your Python script needs to run system commands (kubectl, docker, aws cli, git). Instead of copying the entire command's code into Python, you just call the command from Python and capture what it outputs. It is like telling a helper "run this command and give me the result."
subprocess.run(): execute a command, wait for completion, return result. subprocess.Popen(): advanced version with more control (streams, async). Exit codes: 0 = success, non-zero = failure. stdout/stderr: normal output and error messages.
Never pass user input directly to shell commands. Always split commands into a list: subprocess.run(["kubectl", "get", "pods"]) not subprocess.run("kubectl get pods", shell=True). If you must use shell=True, validate/escape user input to prevent shell injection attacks.
import subprocess
import json
# ===== SIMPLE COMMAND EXECUTION =====
# Run command, ignore output
result = subprocess.run(["echo", "Hello"])
print(f"Exit code: {result.returncode}") # 0 if success
# ===== CAPTURE OUTPUT =====
result = subprocess.run(
["ls", "-la"],
capture_output=True, # capture stdout and stderr
text=True # decode bytes to string
)
if result.returncode == 0:
print("Output:")
print(result.stdout)
else:
print("Error:")
print(result.stderr)
# ===== CHECK FOR COMMAND SUCCESS =====
result = subprocess.run(["docker", "ps"], capture_output=True, text=True)
if result.returncode != 0:
print(f"Docker command failed: {result.stderr}")
else:
containers = result.stdout.strip().split("\n")
print(f"Found {len(containers)} containers")
# ===== CAPTURE JSON OUTPUT =====
# Common pattern: run command that outputs JSON, parse it
result = subprocess.run(
["kubectl", "get", "pods", "-o", "json"],
capture_output=True,
text=True,
check=True # raise CalledProcessError if returncode != 0
)
pods = json.loads(result.stdout)
for item in pods.get("items", []):
print(f"Pod: {item['metadata']['name']}")
# ===== PASS INPUT TO COMMAND =====
# Some commands read from stdin
result = subprocess.run(
["sort"],
input="zebra\napple\nbanana\n",
capture_output=True,
text=True
)
print(result.stdout) # sorted output
# ===== ENVIRONMENT VARIABLES =====
import os
# Pass environment variables to subprocess
env = os.environ.copy()
env["KUBECONFIG"] = "/home/user/.kube/custom.config"
result = subprocess.run(
["kubectl", "get", "nodes"],
capture_output=True,
text=True,
env=env
)
# ===== WORKING DIRECTORY =====
result = subprocess.run(
["git", "log", "--oneline", "-1"],
cwd="/path/to/repo", # run in this directory
capture_output=True,
text=True
)
# ===== TIMEOUT: PREVENT HANGING =====
try:
result = subprocess.run(
["sudo", "systemctl", "restart", "nginx"],
capture_output=True,
text=True,
timeout=30 # abort if command takes >30 seconds
)
except subprocess.TimeoutExpired:
print("Command timed out after 30 seconds")
# ===== ERROR HANDLING =====
def safe_run_command(cmd, timeout=30):
"""Run command safely with full error context."""
try:
result = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=timeout,
check=True # raise on non-zero exit
)
return result.stdout
except subprocess.CalledProcessError as e:
print(f"Command failed with exit code {e.returncode}")
print(f"stdout: {e.stdout}")
print(f"stderr: {e.stderr}")
raise
except subprocess.TimeoutExpired:
print(f"Command timed out after {timeout} seconds")
raise
except FileNotFoundError:
print(f"Command not found (not in PATH)")
raise
# ===== REAL-WORLD EXAMPLE: DEPLOY WITH KUBECTL =====
def apply_kubernetes_manifest(manifest_path):
"""Apply Kubernetes manifest and check status."""
try:
# Apply the manifest
result = subprocess.run(
["kubectl", "apply", "-f", manifest_path],
capture_output=True,
text=True,
check=True
)
print(f"Applied: {result.stdout}")
# Wait for rollout
result = subprocess.run(
["kubectl", "rollout", "status", "deployment/my-app", "-w"],
capture_output=True,
text=True,
timeout=300
)
print("Rollout complete")
return True
except subprocess.CalledProcessError as e:
print(f"Deployment failed: {e.stderr}")
return False
# ===== PARSING STRUCTURED OUTPUT =====
# kubectl get pods outputs as text; convert to JSON for parsing
result = subprocess.run(
["kubectl", "get", "pods", "-o", "json"],
capture_output=True,
text=True,
check=True
)
pods_data = json.loads(result.stdout)
running_pods = [
item["metadata"]["name"]
for item in pods_data["items"]
if item["status"]["phase"] == "Running"
]
print(f"Running pods: {running_pods}")
# ===== MULTIPLE COMMANDS (PIPING) =====
# In shell: ps aux | grep python | wc -l
# In Python: call each command separately, pipe output
# Get process info
ps_result = subprocess.run(
["ps", "aux"],
capture_output=True,
text=True,
check=True
)
# Filter for python
python_procs = [
line for line in ps_result.stdout.split("\n")
if "python" in line.lower()
]
print(f"Found {len(python_procs)} Python processes")
# ===== BACKGROUND PROCESSES (Popen) =====
# For long-running commands, use Popen
process = subprocess.Popen(
["tail", "-f", "/var/log/app.log"],
stdout=subprocess.PIPE,
text=True
)
# Read output line by line
try:
for line in process.stdout:
print(f"Log: {line.strip()}")
if "ERROR" in line:
process.terminate() # stop if error found
break
except KeyboardInterrupt:
process.terminate()
process.wait() # wait for process to exit
# ===== CHECK IF COMMAND EXISTS =====
def command_exists(cmd):
"""Check if command is available in PATH."""
result = subprocess.run(
["which", cmd],
capture_output=True
)
return result.returncode == 0
if command_exists("kubectl"):
print("kubectl is installed")
else:
print("kubectl not found, install it first")
A CI/CD script runs: `git commit`, captures the commit hash, passes it to `docker build --tag myapp:HASH`, then `docker push`, then `kubectl set image deployment/app app=myapp:HASH`. Each command's output feeds into the next. Errors at any stage stop execution and report which step failed.
Write a script that: (1) lists all Docker containers, (2) parses the output to get container IDs, (3) for each container, runs `docker inspect CONTAINER_ID` and extracts the image name, (4) prints a summary of container:image mappings. Handle errors gracefully.
Problem: command works in terminal but fails when run from Python with "command not found".
It raises CalledProcessError if the command exits with a non-zero code. Without check=True, you must manually check result.returncode. Using check=True makes error handling cleaner with try/except.
It captures stdout and stderr so you can read the command's output in your Python script. Without it, output goes to your terminal. You typically want to capture output to parse it or log it.
text=True decodes bytes to strings automatically. text=False (default) returns bytes. If you want to parse output as strings (split lines, search text), use text=True. If binary format, use text=False.
Use try/except for TimeoutExpired. Example: try: result = subprocess.run(["kubectl", "get", "pods"], capture_output=True, timeout=10, check=True). except subprocess.TimeoutExpired: print("Command timed out"). The timeout parameter aborts if the process exceeds the limit.
Every DevOps automation tool calls system commands: deploying with kubectl, building with docker, provisioning with terraform, managing infrastructure with cloud CLIs. subprocess is the bridge between orchestration logic and actual infrastructure management.
subprocess.run() is the modern way to execute commands. Use capture_output=True to get output, text=True to decode as strings. Check exit codes with returncode or use check=True to raise on failure. Always use list form (["cmd", "arg"]) not shell strings to prevent injection. Use timeout to prevent hangs. Parse JSON output for structured data. Error handling with try/except for CalledProcessError, TimeoutExpired, FileNotFoundError. These patterns are core to DevOps automation.