Python Session 05: Subprocess

import subprocess

result = subprocess.run(['ls', '-la', '/tmp'], capture_output=True, text=True)
print(result.stdout)
print(f"Return code: {result.returncode}")

import subprocess

result = subprocess.run(['ping', '-c', '1', 'localhost'], capture_output=True, text=True)

if result.returncode == 0:
    print("Host is reachable")
else:
    print("Host unreachable")
    print(result.stderr)

import subprocess

try:
    result = subprocess.run(['ls', '/nonexistent'], capture_output=True, text=True, check=True)
except subprocess.CalledProcessError as e:
    print(f"Command failed: {e}")
    print(f"stderr: {e.stderr}")

import subprocess

# Note: shell=True has security implications - don't use with untrusted input
result = subprocess.run('ps aux | grep python | head -5', shell=True, capture_output=True, text=True)
print(result.stdout)

import subprocess

# Safer: pipe between processes
ps = subprocess.Popen(['ps', 'aux'], stdout=subprocess.PIPE)
grep = subprocess.Popen(['grep', 'python'], stdin=ps.stdout, stdout=subprocess.PIPE)
ps.stdout.close()
output = grep.communicate()[0]
print(output.decode())

import os

print(os.environ.get('HOME'))
print(os.environ.get('VAULT_ADDR', 'not set'))

# All environment
for key, value in os.environ.items():
    if 'PATH' in key:
        print(f"{key}: {value[:50]}...")

import subprocess
import os

my_env = os.environ.copy()
my_env['MY_VAR'] = 'my_value'

result = subprocess.run(['printenv', 'MY_VAR'], capture_output=True, text=True, env=my_env)
print(result.stdout)  # my_value

import os

path = os.path.expandvars('$HOME/.ssh/config')
print(path)  # /home/user/.ssh/config

user_home = os.path.expanduser('~')
print(user_home)  # /home/user

import subprocess
import json

result = subprocess.run(['kubectl', 'get', 'pods', '-o', 'json'], capture_output=True, text=True)

if result.returncode == 0:
    data = json.loads(result.stdout)
    for pod in data.get('items', []):
        print(pod['metadata']['name'])

import subprocess

try:
    result = subprocess.run(['sleep', '10'], timeout=2, capture_output=True)
except subprocess.TimeoutExpired:
    print("Command timed out")

import subprocess

result = subprocess.run(['pwd'], capture_output=True, text=True, cwd='/tmp')
print(result.stdout.strip())  # /tmp

result = subprocess.run(['git', 'status'], capture_output=True, text=True,
                       cwd='/home/user/project')

import subprocess
import shlex

host = "10.50.1.110"
cmd = ['ssh', '-o', 'ConnectTimeout=5', host, 'uptime']

result = subprocess.run(cmd, capture_output=True, text=True)
print(result.stdout)

# If you have a string command, use shlex.split
cmd_str = "ls -la /tmp"
cmd_list = shlex.split(cmd_str)
print(cmd_list)  # ['ls', '-la', '/tmp']