Python subprocess.run() Deep Dive: Isolation, Timeouts, and Captured Output
subprocess.run() is one of those functions you think you understand until you actually need it to do something precise.
Here's everything it does โ from basic execution to full sandboxing โ with working examples for each pattern.
๐ Free: AI Publishing Checklist โ 7 steps in Python ยท Full pipeline: germy5.gumroad.com/l/xhxkzz (pay what you want, min $9.99)
The signature
subprocess.run(
args, # command as list or string
*,
stdin=None,
input=None,
capture_output=False, # shorthand for stdout=PIPE, stderr=PIPE
stdout=None,
stderr=None,
shell=False,
cwd=None, # working directory for the subprocess
timeout=None, # seconds before TimeoutExpired is raised
check=False, # raise CalledProcessError on non-zero exit
encoding=None,
errors=None,
env=None, # environment variables dict
text=False, # alias for encoding='utf-8'
**kwargs
) -> CompletedProcess
Pattern 1: Capture stdout and stderr separately
import subprocess
result = subprocess.run(
["python3", "-c", "import sys; print('out'); print('err', file=sys.stderr)"],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
)
print(f"stdout: {result.stdout!r}") # 'out\n'
print(f"stderr: {result.stderr!r}") # 'err\n'
print(f"exit: {result.returncode}") # 0
Use capture_output=True as shorthand for stdout=PIPE, stderr=PIPE:
result = subprocess.run(
["python3", "-c", "print('hello')"],
capture_output=True,
text=True,
)
print(result.stdout) # 'hello\n'
When to use: Any time you need to inspect the output programmatically rather than let it print to the terminal.
Pattern 2: Timeout enforcement
import subprocess
def run_with_timeout(cmd: list, timeout_seconds: int = 30) -> subprocess.CompletedProcess:
try:
return subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=timeout_seconds,
)
except subprocess.TimeoutExpired as e:
print(f"Command timed out after {timeout_seconds}s")
print(f"Partial stdout: {e.stdout}")
raise
# This will time out:
try:
run_with_timeout(
["python3", "-c", "import time; time.sleep(100)"],
timeout_seconds=2,
)
except subprocess.TimeoutExpired:
print("Caught timeout โ subprocess killed")
When to use: Any time you're running untrusted or user-supplied code. Without a timeout, an infinite loop hangs your process forever.
Pattern 3: Directory isolation with cwd
import subprocess, tempfile, os
def run_isolated(script_content: str, timeout: int = 30) -> subprocess.CompletedProcess:
"""Run a script in a fresh temp directory โ no access to parent dir."""
with tempfile.TemporaryDirectory() as tmpdir:
script_path = os.path.join(tmpdir, "script.py")
with open(script_path, "w") as f:
f.write(script_content)
return subprocess.run(
["python3", script_path],
capture_output=True,
text=True,
timeout=timeout,
cwd=tmpdir, # subprocess starts in the temp dir
)
# Script that tries to read a parent directory file โ will fail cleanly
result = run_isolated("""
import os
try:
files = os.listdir('..')
print(f"Parent dir: {files[:3]}")
except PermissionError:
print("No access to parent")
# But can write/read within its own dir
with open('output.txt', 'w') as f:
f.write('hello')
print(f"Created file: {os.path.exists('output.txt')}")
""")
print(result.stdout)
# Created file: True (but the temp dir is deleted when context exits)
When to use: Running generated or untrusted code where you want a clean working directory and no accidental access to your project files.
Pattern 4: Environment variable injection
import subprocess, os
def run_with_env(cmd: list, extra_env: dict) -> subprocess.CompletedProcess:
"""Run with a modified environment โ inherits current env, adds/overrides keys."""
env = os.environ.copy()
env.update(extra_env)
return subprocess.run(cmd, capture_output=True, text=True, env=env)
result = run_with_env(
["python3", "-c", "import os; print(os.environ.get('MY_VAR', 'not set'))"],
{"MY_VAR": "injected_value"},
)
print(result.stdout) # injected_value
For a completely clean environment (no inherited vars):
result = subprocess.run(
["python3", "-c", "import os; print(len(os.environ))"],
capture_output=True,
text=True,
env={"PATH": "/usr/bin:/bin"}, # minimal env only
)
# Output: 1
Pattern 5: check=True for fail-fast pipelines
import subprocess
def run_checked(cmd: list) -> str:
"""Run command, raise immediately on non-zero exit."""
result = subprocess.run(
cmd,
capture_output=True,
text=True,
check=True, # raises CalledProcessError on non-zero exit
)
return result.stdout
try:
output = run_checked(["python3", "-c", "exit(1)"])
except subprocess.CalledProcessError as e:
print(f"Command failed with exit code {e.returncode}")
print(f"stderr: {e.stderr}")
When to use: In pipelines where a failure at any step should stop the whole chain. Equivalent to set -e in bash.
Pattern 6: stdin piping
import subprocess
result = subprocess.run(
["python3", "-c", "import sys; data = sys.stdin.read(); print(data.upper())"],
input="hello from stdin\n",
capture_output=True,
text=True,
)
print(result.stdout) # HELLO FROM STDIN
Pipe one process's output to another:
p1 = subprocess.run(
["python3", "-c", "print('line1\\nline2\\nline3')"],
capture_output=True, text=True,
)
p2 = subprocess.run(
["python3", "-c", "import sys; [print(l.upper()) for l in sys.stdin]"],
input=p1.stdout,
capture_output=True, text=True,
)
print(p2.stdout)
# LINE1
# LINE2
# LINE3
Full validation harness
This is what I use in the ebook pipeline โ all patterns combined:
import subprocess, tempfile, os
from dataclasses import dataclass
@dataclass
class ValidationResult:
"""Result of running a script through the validation harness."""
passed: bool
exit_code: int
stdout: str
stderr: str
timed_out: bool = False
def __bool__(self):
return self.passed
def validate_script(
code: str,
timeout: int = 30,
extra_env: dict = None,
) -> ValidationResult:
"""
Run a Python script in complete isolation and return structured results.
Args:
code: Python source code to run
timeout: Maximum execution time in seconds
extra_env: Additional environment variables to inject
Returns:
ValidationResult with pass/fail and captured output
"""
env = {"PATH": os.environ.get("PATH", "/usr/bin:/bin")}
if extra_env:
env.update(extra_env)
with tempfile.TemporaryDirectory() as tmpdir:
script_path = os.path.join(tmpdir, "test.py")
with open(script_path, "w", encoding="utf-8") as f:
f.write(code)
try:
result = subprocess.run(
["python3", script_path],
capture_output=True,
text=True,
timeout=timeout,
cwd=tmpdir,
env=env,
)
return ValidationResult(
passed=(result.returncode == 0),
exit_code=result.returncode,
stdout=result.stdout,
stderr=result.stderr,
)
except subprocess.TimeoutExpired:
return ValidationResult(
passed=False,
exit_code=-1,
stdout="",
stderr=f"Timed out after {timeout}s",
timed_out=True,
)
# Usage
code = '''
def greet(name: str) -> str:
"""Return a greeting string."""
return f"Hello, {name}!"
if __name__ == "__main__":
print(greet("world"))
'''
result = validate_script(code)
if result:
print(f"โ
Passed โ stdout: {result.stdout.strip()}")
else:
print(f"โ Failed โ stderr: {result.stderr}")
CompletedProcess reference
result = subprocess.run(["echo", "hello"], capture_output=True, text=True)
result.args # ['echo', 'hello']
result.returncode # 0
result.stdout # 'hello\n'
result.stderr # ''
result.check_returncode() # raises CalledProcessError if non-zero
This validation harness (and the full two-gate system using ast.parse() + subprocess.run()) is included in: germy5.gumroad.com/l/xhxkzz โ pay what you want, min $9.99.
Top comments (0)