Python subprocess: Running External Commands

import subprocess
 
# Simple command
result = subprocess.run(["ls", "-la"])
print(result.returncode)  # 0 if successful
 
# Capture output
result = subprocess.run(
    ["ls", "-la"],
    capture_output=True,
    text=True  # Return strings, not bytes
)
print(result.stdout)
print(result.stderr)
 
# Check for errors
result = subprocess.run(["ls", "/nonexistent"], capture_output=True, text=True)
if result.returncode != 0:
    print(f"Error: {result.stderr}")

import subprocess
 
# Raise exception on failure
try:
    subprocess.run(["ls", "/nonexistent"], check=True, capture_output=True, text=True)
except subprocess.CalledProcessError as e:
    print(f"Command failed with code {e.returncode}")
    print(f"stderr: {e.stderr}")

import subprocess
 
# Send input to command
result = subprocess.run(
    ["grep", "hello"],
    input="hello world\ngoodbye world\nhello again",
    capture_output=True,
    text=True
)
print(result.stdout)  # hello world\nhello again
 
# Pipe to file
with open("output.txt", "w") as f:
    subprocess.run(["ls", "-la"], stdout=f)
 
# Discard output
subprocess.run(["noisy-command"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)

import subprocess
 
try:
    result = subprocess.run(
        ["sleep", "10"],
        timeout=2
    )
except subprocess.TimeoutExpired:
    print("Command timed out")
 
# With capture
try:
    result = subprocess.run(
        ["long-command"],
        capture_output=True,
        text=True,
        timeout=30
    )
except subprocess.TimeoutExpired as e:
    print(f"Timed out after {e.timeout}s")
    # e.stdout and e.stderr may have partial output

import subprocess
 
# AVOID: shell=True with user input (security risk!)
# subprocess.run(f"ls {user_input}", shell=True)  # DANGEROUS
 
# OK for fixed commands
result = subprocess.run(
    "ls -la | grep py",
    shell=True,
    capture_output=True,
    text=True
)
 
# Better: use Python for pipes (see below)

import subprocess
 
# Python-native piping
p1 = subprocess.Popen(
    ["ls", "-la"],
    stdout=subprocess.PIPE
)
p2 = subprocess.Popen(
    ["grep", "py"],
    stdin=p1.stdout,
    stdout=subprocess.PIPE,
    text=True
)
p1.stdout.close()  # Allow p1 to receive SIGPIPE
output = p2.communicate()[0]
print(output)
 
# Simpler with shell (for trusted commands)
result = subprocess.run(
    "ls -la | grep py | wc -l",
    shell=True,
    capture_output=True,
    text=True
)

import subprocess
 
# Start process without waiting
proc = subprocess.Popen(
    ["long-running-command"],
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    text=True
)
 
# Do other work...
 
# Wait and get output
stdout, stderr = proc.communicate()
print(f"Return code: {proc.returncode}")
 
# Poll without blocking
while proc.poll() is None:
    print("Still running...")
    # Do other work

import subprocess
 
# Read output line by line
proc = subprocess.Popen(
    ["tail", "-f", "logfile.log"],
    stdout=subprocess.PIPE,
    text=True
)
 
for line in proc.stdout:
    print(f"Got: {line.strip()}")
    if "error" in line.lower():
        proc.terminate()
        break

import subprocess
import os
 
# Pass custom environment
env = os.environ.copy()
env["MY_VAR"] = "my_value"
 
result = subprocess.run(
    ["printenv", "MY_VAR"],
    env=env,
    capture_output=True,
    text=True
)
 
# Clear environment (be careful)
result = subprocess.run(
    ["command"],
    env={"PATH": "/usr/bin"}
)

import subprocess
 
# Run in specific directory
result = subprocess.run(
    ["git", "status"],
    cwd="/path/to/repo",
    capture_output=True,
    text=True
)

import subprocess
import shlex
 
# Parse command string safely
cmd = "ls -la 'my file.txt'"
args = shlex.split(cmd)  # ['ls', '-la', 'my file.txt']
subprocess.run(args)
 
# Quote for shell
filename = "file with spaces.txt"
safe = shlex.quote(filename)  # "'file with spaces.txt'"
 
# Never do this with user input!
# subprocess.run(f"rm {user_input}", shell=True)
 
# Do this instead
subprocess.run(["rm", user_input])  # Arguments are properly escaped

import subprocess
 
def run_command(cmd, **kwargs):
    """Run command with sensible defaults."""
    defaults = {
        "capture_output": True,
        "text": True,
        "check": True,
    }
    defaults.update(kwargs)
    return subprocess.run(cmd, **defaults)
 
def get_output(cmd):
    """Get stdout from command."""
    result = subprocess.run(cmd, capture_output=True, text=True, check=True)
    return result.stdout.strip()
 
def command_exists(name):
    """Check if command is available."""
    try:
        subprocess.run(
            ["which", name],
            capture_output=True,
            check=True
        )
        return True
    except subprocess.CalledProcessError:
        return False
 
def run_with_retry(cmd, retries=3, delay=1):
    """Retry command on failure."""
    import time
    for attempt in range(retries):
        try:
            return subprocess.run(cmd, check=True, capture_output=True, text=True)
        except subprocess.CalledProcessError:
            if attempt < retries - 1:
                time.sleep(delay)
    raise

import subprocess
 
def git_status(repo_path):
    result = subprocess.run(
        ["git", "status", "--porcelain"],
        cwd=repo_path,
        capture_output=True,
        text=True,
        check=True
    )
    return result.stdout.strip()
 
def git_commit(repo_path, message):
    subprocess.run(
        ["git", "add", "-A"],
        cwd=repo_path,
        check=True
    )
    subprocess.run(
        ["git", "commit", "-m", message],
        cwd=repo_path,
        check=True
    )
 
def git_current_branch(repo_path):
    result = subprocess.run(
        ["git", "branch", "--show-current"],
        cwd=repo_path,
        capture_output=True,
        text=True,
        check=True
    )
    return result.stdout.strip()

import asyncio
 
async def run_async(cmd):
    proc = await asyncio.create_subprocess_exec(
        *cmd,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE
    )
    stdout, stderr = await proc.communicate()
    return stdout.decode(), stderr.decode(), proc.returncode
 
# Run multiple commands concurrently
async def main():
    results = await asyncio.gather(
        run_async(["command1"]),
        run_async(["command2"]),
        run_async(["command3"]),
    )
    for stdout, stderr, code in results:
        print(f"Code: {code}, Output: {stdout[:50]}")
 
asyncio.run(main())

import subprocess
 
def safe_run(cmd, **kwargs):
    """Run command with comprehensive error handling."""
    try:
        return subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            check=True,
            timeout=30,
            **kwargs
        )
    except subprocess.CalledProcessError as e:
        print(f"Command failed: {e.cmd}")
        print(f"Return code: {e.returncode}")
        print(f"stderr: {e.stderr}")
        raise
    except subprocess.TimeoutExpired as e:
        print(f"Command timed out after {e.timeout}s")
        raise
    except FileNotFoundError:
        print(f"Command not found: {cmd[0]}")
        raise

# Always use list form when possible
subprocess.run(["ls", "-la"])  # Good
subprocess.run("ls -la", shell=True)  # Avoid
 
# Always capture output for commands that might fail
result = subprocess.run(cmd, capture_output=True, text=True)
 
# Use check=True unless you handle errors manually
subprocess.run(cmd, check=True)
 
# Set timeouts for external commands
subprocess.run(cmd, timeout=30)
 
# Use text=True for string output
subprocess.run(cmd, capture_output=True, text=True)