← Home

135. Subprocess

Running shell commands from Python with subprocess

135. Subprocess

Aryan’s first Python attempt reads shell output directly.

πŸ€” Socratic question: You know ps aux works in the terminal. Can’t you just… run that from Python? Yes. But should you? What happens on Windows? What if the output format changes? This chapter is about learning why the obvious approach fails.

He knows ps aux works. His first instinct: run it from Python.

import subprocess

result = subprocess.run(
    ['ps', 'aux'],
    capture_output=True, text=True
)
lines = result.stdout.splitlines()
# Sort by RSS (column index 5 on Linux)
processes = sorted(lines[1:], key=lambda l: int(l.split()[5] or 0), reverse=True)
for line in processes[:10]:
    parts = line.split(None, 10)
    print(f"PID {parts[1]:>6}  RSS {int(parts[5])//1024:>5} MB  {parts[10][:30]}")

It works. On his Mac. On Linux the column indices are slightly different. On Windows, ps doesn’t exist at all. And when a process exits mid-run, he gets a ValueError trying to parse the garbled row.

He also notices that subprocess.run() without check=True silently returns exit code 1 on failure. Without capture_output=True the output goes straight to the terminal instead of into result.stdout.

He uses it enough to understand it. But he knows there’s a better way.

πŸ’‘ Real-world: DevOps engineers still use subprocess all the time β€” for running git commands, calling system utilities, orchestrating shell scripts. It’s not wrong, it’s just the wrong tool for this job. Every tool in your belt has its place.

πŸ’‘ Fun fact: subprocess replaced the older os.system(), os.popen(), and the commands module β€” all in one PEP (PEP 324, accepted in 2003). The designers wanted a single, secure, cross-platform way to run processes. os.system() is still in Python for backward compatibility, but every style guide since 2003 says: use subprocess.

⚠️ Watch out: Without capture_output=True, result.stdout is None β€” not an empty string, not an empty bytes object, literally None. Calling .split('\n') on None gives you AttributeError: 'NoneType' object has no attribute 'split'. Always add capture_output=True and text=True together.

πŸ€” Think about it: check=True raises CalledProcessError when the command exits with a non-zero code. Without it, a failed command looks like a successful one β€” you just get empty output. What kinds of silent bugs could that cause in production? Why is failing loudly usually better than failing silently?


🐚 Aryan wants to call ps and free from Python before he knows about psutil. His subprocess calls silently discard output, use the wrong flag for text mode, and crash on non-zero exit codes without a useful message.

Learning objectives

  • Use capture_output=True to capture stdout/stderr
  • Use text=True to get str instead of bytes
  • Use check=True to raise on non-zero exit codes
  • Handle CalledProcessError and FileNotFoundError
  • Prefer subprocess.run() over os.system()

Key concepts

  • subprocess.run() β€” run a command, wait for it
  • capture_output=True β€” capture stdout/stderr
  • text=True β€” decode bytes to str
  • check=True β€” raise CalledProcessError on failure
  • CalledProcessError β€” non-zero exit code exception

Try it

Concept detail

Running Shell Commands with subprocess

subprocess is the stdlib module for running external programs from Python.

The right way β€” subprocess.run()

import subprocess

result = subprocess.run(
    ['ps', 'aux', '--sort=-%mem'],
    capture_output=True,   # capture stdout/stderr
    text=True,             # return str, not bytes
    check=True             # raise on non-zero exit
)
print(result.stdout)

Common flags

capture_output=True   # same as stdout=PIPE, stderr=PIPE
text=True             # decode bytes to str (uses locale encoding)
check=True            # raise CalledProcessError if exit code != 0
timeout=10            # raise TimeoutExpired after N seconds

check_output() shorthand

# Returns stdout as str, raises on failure
output = subprocess.check_output(['free', '-h'], text=True)

Error handling

try:
    result = subprocess.run(['ps', 'aux'], capture_output=True, text=True, check=True)
except subprocess.CalledProcessError as e:
    print(f'Command failed (exit {e.returncode}): {e.stderr}')
except FileNotFoundError:
    print('Command not found β€” not available on this OS')

Why subprocess beats os.system()

# os.system β€” output goes to terminal, no capture, no error info
os.system('ps aux')   # bad

# subprocess β€” full control
result = subprocess.run(['ps', 'aux'], capture_output=True, text=True)

Shell commands for RAM monitoring

free -h                          # total/used/free RAM
ps aux --sort=-%mem | head -10   # top 10 by memory
pgrep -a chrome                  # PIDs + cmd for chrome

Solution

import subprocess

def get_ram_usage() -> str:
    result = subprocess.run(
        ['free', '-h'],
        capture_output=True,   # capture stdout + stderr
        text=True              # decode bytes β†’ str automatically
    )
    return result.stdout

def get_top_processes(n: int = 5) -> str:
    result = subprocess.run(
        ['ps', 'aux', '--sort=-%mem'],
        capture_output=True,
        text=True
    )
    lines = result.stdout.split('\n')
    return '\n'.join(lines[:n + 1])

def run_command(cmd: list[str]) -> str:
    # check=True raises CalledProcessError on non-zero exit
    result = subprocess.run(cmd, capture_output=True, text=True, check=True)
    return result.stdout

Tests

import subprocess
import inspect
import pytest
from unittest.mock import patch, MagicMock

FREE_OUTPUT = (
    '              total        used        free      shared  buff/cache   available\n'
    'Mem:          15987        8234        2156         512        5597        6841\n'
    'Swap:          2047           0        2047\n'
)
PS_OUTPUT = (
    'USER   PID %CPU %MEM    VSZ    RSS TTY  STAT START   TIME COMMAND\n'
    'aryan  812  2.1 12.4 6291456 2048000 ?  Sl   10:23   0:45 chrome\n'
    'aryan  421  0.5  4.2 1048576  671000 ?  S    09:10   0:12 python3\n'
)

def _mock_run(cmd, **kwargs):
    result = MagicMock()
    result.returncode = 0
    result.stdout = FREE_OUTPUT if 'free' in cmd else PS_OUTPUT
    result.stderr = ''
    return result

def test_get_ram_usage_returns_string():
    with patch('subprocess.run', side_effect=_mock_run):
        result = get_ram_usage()
    assert isinstance(result, str)
    assert len(result) > 0

def test_get_top_processes_returns_string():
    with patch('subprocess.run', side_effect=_mock_run):
        result = get_top_processes(3)
    assert isinstance(result, str)

def test_run_command_captures_output():
    fake = MagicMock()
    fake.stdout = 'hello ram manager\n'
    fake.returncode = 0
    with patch('subprocess.run', return_value=fake):
        result = run_command(['echo', 'hello ram manager'])
    assert 'hello ram manager' in result

def test_run_command_raises_on_failure():
    with patch('subprocess.run', side_effect=subprocess.CalledProcessError(1, ['false'])):
        with pytest.raises(subprocess.CalledProcessError):
            run_command(['false'])

def test_get_ram_usage_uses_capture_output():
    src = inspect.getsource(get_ram_usage)
    assert 'capture_output' in src, 'Must use capture_output=True'

def test_get_top_processes_uses_text_mode():
    src = inspect.getsource(get_top_processes)
    assert 'text=True' in src, 'Must use text=True to get str not bytes'

def test_run_command_uses_check_true():
    src = inspect.getsource(run_command)
    assert 'check=True' in src, 'Must use check=True to raise CalledProcessError on failure'

Resources