subprocess is Python’s stdlib module for running shell commands. import subprocess. run: r = subprocess.run(["ls","-la"], capture_output=True, text=True); r.stdout; r.returncode. check: subprocess.run(["pytest"], check=True) — raises CalledProcessError on non-zero exit. check_output: out = subprocess.check_output(["git","rev-parse","HEAD"], text=True).strip(). Shell string: subprocess.run("ls -la | grep py", shell=True, check=True). PIPE: r = subprocess.run(["cat","file.txt"], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True). timeout: subprocess.run(["sleep","10"], timeout=5) — raises TimeoutExpired. env: subprocess.run(["make"], env={**os.environ,"DEBUG":"1"}). cwd: subprocess.run(["npm","install"], cwd="/app"). Popen: proc = subprocess.Popen(["tail","-f","app.log"], stdout=subprocess.PIPE, text=True). iter lines: for line in proc.stdout: print(line,end=""). communicate: stdout, stderr = proc.communicate(input="data\n"). wait: proc.wait(). kill: proc.kill(). STDOUT: subprocess.run(["cmd"], stdout=subprocess.PIPE, stderr=subprocess.STDOUT) — merge. shlex.split: import shlex; args = shlex.split("git commit -m 'hello world'"). CalledProcessError: except subprocess.CalledProcessError as e: e.returncode; e.stdout; e.stderr. CompletedProcess: r.args; r.returncode; r.stdout; r.stderr. Claude Code generates subprocess wrappers, build runners, Git helpers, and streaming log monitors.
CLAUDE.md for subprocess
## subprocess Stack
- Stdlib: import subprocess, shlex, os
- Run: subprocess.run(["cmd","arg"], capture_output=True, text=True, check=True, timeout=N)
- Capture: r.stdout.strip() | r.returncode | r.stderr
- Stream: Popen(cmd, stdout=PIPE, text=True) | for line in proc.stdout: ...
- Error: except subprocess.CalledProcessError as e: print(e.returncode, e.stderr)
- Safety: NEVER pass shell=True with user input — use list args insteadsubprocess Command Pipeline
# app/shell.py — subprocess run, stream, Popen, Git, build, timeout, env helpers
from __future__ import annotations
import io
import os
import shlex
import subprocess
import sys
import tempfile
import time
import logging
from contextlib import contextmanager
from dataclasses import dataclass
from pathlib import Path
from typing import Generator, Iterator
log = logging.getLogger(__name__)
# ─────────────────────────────────────────────────────────────────────────────
# 1. Core run helpers
# ─────────────────────────────────────────────────────────────────────────────
@dataclass
class RunResult:
args: list[str]
returncode: int
stdout: str
stderr: str
elapsed: float
@property
def ok(self) -> bool:
return self.returncode == 0
def __str__(self) -> str:
return self.stdout.strip()
def run(
cmd: str | list[str],
cwd: str | Path | None = None,
env: dict[str, str] | None = None,
timeout: float | None = None,
check: bool = True,
input: str | None = None,
extra_env: dict[str, str] | None = None,
) -> RunResult:
"""
Run a command and return a RunResult.
Accepts list or string args (string is split with shlex).
Never uses shell=True — avoids injection vulnerabilities.
Example:
r = run(["git", "rev-parse", "HEAD"], cwd="/repo")
print(r.stdout) # commit hash
r = run("pytest -v --tb=short", check=False)
if not r.ok:
print(r.stderr)
"""
if isinstance(cmd, str):
args = shlex.split(cmd)
else:
args = list(cmd)
merged_env = dict(os.environ)
if env:
merged_env = dict(env)
if extra_env:
merged_env.update(extra_env)
t0 = time.perf_counter()
try:
result = subprocess.run(
args,
capture_output=True,
text=True,
cwd=cwd,
env=merged_env,
timeout=timeout,
input=input,
)
except subprocess.TimeoutExpired as exc:
raise TimeoutError(
f"Command timed out after {timeout}s: {' '.join(args)}"
) from exc
elapsed = time.perf_counter() - t0
run_result = RunResult(
args=args,
returncode=result.returncode,
stdout=result.stdout,
stderr=result.stderr,
elapsed=elapsed,
)
if check and result.returncode != 0:
raise subprocess.CalledProcessError(
result.returncode, args,
output=result.stdout, stderr=result.stderr,
)
log.debug("run %r → exit=%d (%.2fs)", args, result.returncode, elapsed)
return run_result
def run_silent(cmd: str | list[str], **kwargs) -> bool:
"""
Run a command; return True if exit code is 0, False otherwise.
Never raises on error.
Example:
if run_silent("which docker"):
print("Docker is installed")
"""
try:
run(cmd, check=True, **kwargs)
return True
except (subprocess.CalledProcessError, FileNotFoundError, TimeoutError):
return False
def capture(cmd: str | list[str], strip: bool = True, **kwargs) -> str:
"""
Run a command and return its stdout as a string.
Example:
branch = capture("git rev-parse --abbrev-ref HEAD").strip()
version = capture(["python", "--version"])
"""
result = run(cmd, check=True, **kwargs)
return result.stdout.strip() if strip else result.stdout
def capture_lines(cmd: str | list[str], **kwargs) -> list[str]:
"""
Run a command and return non-empty lines of stdout.
Example:
files = capture_lines("git diff --name-only HEAD~1")
"""
output = capture(cmd, strip=False, **kwargs)
return [line for line in output.splitlines() if line.strip()]
# ─────────────────────────────────────────────────────────────────────────────
# 2. Streaming helpers
# ─────────────────────────────────────────────────────────────────────────────
def stream_lines(
cmd: str | list[str],
cwd: str | Path | None = None,
env: dict[str, str] | None = None,
timeout: float | None = None,
merge_stderr: bool = False,
) -> Iterator[str]:
"""
Stream stdout line-by-line from a long-running command.
Example:
for line in stream_lines("tail -f app.log"):
parse_log_line(line)
for line in stream_lines(["pytest", "-v"], merge_stderr=True):
print(line)
"""
if isinstance(cmd, str):
cmd = shlex.split(cmd)
stderr_dest = subprocess.STDOUT if merge_stderr else subprocess.PIPE
with subprocess.Popen(
cmd,
stdout=subprocess.PIPE,
stderr=stderr_dest,
text=True,
cwd=cwd,
env=env,
bufsize=1,
) as proc:
deadline = time.monotonic() + timeout if timeout else None
for line in proc.stdout: # type: ignore[union-attr]
if deadline and time.monotonic() > deadline:
proc.kill()
raise TimeoutError(f"Command timed out: {' '.join(cmd)}")
yield line.rstrip("\n")
def run_and_log(
cmd: str | list[str],
logger=None,
level: int = logging.INFO,
**kwargs,
) -> RunResult:
"""
Run a command, streaming each output line to a logger.
Example:
run_and_log("npm run build", logger=log)
"""
_log = logger or log
lines: list[str] = []
for line in stream_lines(cmd, merge_stderr=True, **kwargs):
_log.log(level, line)
lines.append(line)
# Re-run for exit code (stream_lines gives no returncode)
return run(cmd, check=False, **kwargs)
# ─────────────────────────────────────────────────────────────────────────────
# 3. Context helpers
# ─────────────────────────────────────────────────────────────────────────────
@contextmanager
def in_dir(path: str | Path) -> Generator[Path, None, None]:
"""
Context manager: temporarily change working directory.
Example:
with in_dir("/repo"):
run("make build")
"""
before = Path.cwd()
os.chdir(path)
try:
yield Path(path)
finally:
os.chdir(before)
@contextmanager
def with_env(**env_vars: str) -> Generator[None, None, None]:
"""
Context manager: temporarily set environment variables.
Example:
with with_env(FLASK_ENV="testing", DEBUG="1"):
run("pytest")
"""
original = {k: os.environ.get(k) for k in env_vars}
os.environ.update(env_vars)
try:
yield
finally:
for k, v in original.items():
if v is None:
os.environ.pop(k, None)
else:
os.environ[k] = v
# ─────────────────────────────────────────────────────────────────────────────
# 4. Git helpers
# ─────────────────────────────────────────────────────────────────────────────
class Git:
"""
Convenience wrapper for common git commands.
Example:
git = Git("/path/to/repo")
print(git.current_branch())
print(git.commit_hash())
git.add(".")
git.commit("fix: update config")
git.push()
"""
def __init__(self, repo_dir: str | Path = ".") -> None:
self.repo = Path(repo_dir)
def _run(self, *args: str, check: bool = True) -> RunResult:
return run(["git", *args], cwd=self.repo, check=check)
def current_branch(self) -> str:
return self._run("rev-parse", "--abbrev-ref", "HEAD").stdout.strip()
def commit_hash(self, short: bool = True) -> str:
args = ["rev-parse"]
if short:
args.append("--short")
args.append("HEAD")
return self._run(*args).stdout.strip()
def is_dirty(self) -> bool:
return bool(self._run("status", "--porcelain").stdout.strip())
def changed_files(self, ref: str = "HEAD") -> list[str]:
return capture_lines(["git", "diff", "--name-only", ref], cwd=self.repo)
def log(self, n: int = 10, oneline: bool = True) -> str:
args = ["log", f"-{n}"]
if oneline:
args.append("--oneline")
return self._run(*args).stdout
def add(self, *paths: str) -> None:
self._run("add", *paths)
def commit(self, message: str, allow_empty: bool = False) -> None:
args = ["commit", "-m", message]
if allow_empty:
args.append("--allow-empty")
self._run(*args)
def push(self, remote: str = "origin", branch: str | None = None) -> None:
args = ["push", remote]
if branch:
args.append(branch)
self._run(*args)
def pull(self, remote: str = "origin") -> None:
self._run("pull", remote)
# ─────────────────────────────────────────────────────────────────────────────
# 5. Build / task runner helpers
# ─────────────────────────────────────────────────────────────────────────────
def which(cmd: str) -> str | None:
"""
Return full path of cmd if it exists in PATH, else None.
Example:
if which("docker"):
run("docker build -t myapp .")
"""
import shutil
return shutil.which(cmd)
def require_command(cmd: str) -> None:
"""
Abort with a helpful message if cmd is not installed.
Example:
require_command("ffmpeg")
require_command("node")
"""
if not which(cmd):
raise EnvironmentError(
f"Required command not found: {cmd!r}\n"
f"Install it and make sure it is on your PATH."
)
def pipe(
first_cmd: str | list[str],
second_cmd: str | list[str],
cwd: str | Path | None = None,
) -> str:
"""
Pipe stdout of first_cmd into stdin of second_cmd.
Equivalent to: first_cmd | second_cmd
Example:
output = pipe("cat access.log", "grep 'ERROR'")
output = pipe(["git", "log", "--oneline"], ["head", "-20"])
"""
if isinstance(first_cmd, str):
first_cmd = shlex.split(first_cmd)
if isinstance(second_cmd, str):
second_cmd = shlex.split(second_cmd)
p1 = subprocess.Popen(first_cmd, stdout=subprocess.PIPE, cwd=cwd)
p2 = subprocess.Popen(
second_cmd, stdin=p1.stdout, stdout=subprocess.PIPE,
stderr=subprocess.PIPE, text=True, cwd=cwd,
)
p1.stdout.close() # type: ignore[union-attr]
stdout, _ = p2.communicate()
p1.wait()
return stdout
# ─────────────────────────────────────────────────────────────────────────────
# Demo
# ─────────────────────────────────────────────────────────────────────────────
if __name__ == "__main__":
print("=== subprocess demo ===")
print("\n--- capture ---")
python_version = capture([sys.executable, "--version"])
print(f" Python: {python_version}")
print("\n--- run with check=False ---")
r = run(["ls", "/nonexistent-path-for-demo"], check=False)
print(f" exit={r.returncode}, ok={r.ok}")
print(f" stderr: {r.stderr.strip()!r}")
print("\n--- capture_lines ---")
lines = capture_lines(["ls", str(Path.home())])
print(f" Home dir files (first 5): {lines[:5]}")
print("\n--- stream_lines ---")
count = 0
for line in stream_lines([sys.executable, "-c",
"import time\nfor i in range(3):\n print(f'line {i}')\n import sys; sys.stdout.flush()"]):
print(f" streamed: {line!r}")
count += 1
print(f" total streamed lines: {count}")
print("\n--- run_silent ---")
print(f" 'ls /tmp': {run_silent(['ls', '/tmp'])}")
print(f" 'false': {run_silent(['false'])}")
print("\n--- pipe ---")
out = pipe(["echo", "alpha\nbeta\ngamma"], ["grep", "beta"])
print(f" echo | grep beta: {out.strip()!r}")
print("\n--- which ---")
for tool in ["git", "python3", "definitely-not-installed-xyz"]:
loc = which(tool)
print(f" which({tool!r}): {loc or 'not found'}")
print("\n=== done ===")For the sh alternative — the sh library (PyPI: sh) wraps subprocess with a magic-attribute API (sh.git("status"), sh.ls("-la")), supports background processes, piping, and callbacks with minimal boilerplate; Python’s stdlib subprocess requires explicit list args and is more verbose but has zero dependencies, runs on all platforms (including Windows), and is always available — use sh when you want clean one-liner shell scripting syntax on Unix/macOS, stdlib subprocess when you need Windows compatibility, minimal dependencies, or precise control over streaming and error handling. For the plumbum alternative — plumbum provides a full command pipeline DSL: local["git"]["status"](), cmd1 | cmd2, remote SSH execution, and a CLI framework; subprocess is the stdlib foundation without abstractions — use plumbum for complex shell pipeline DSLs and remote execution in scripts, subprocess when you want explicit, portable code without external dependencies. The Claude Skills 360 bundle includes subprocess skill sets covering run()/run_silent()/capture()/capture_lines() one-shot helpers, stream_lines()/run_and_log() streaming, in_dir()/with_env() context managers, Git wrapper class with add/commit/push/log, which()/require_command()/pipe() utilities. Start with the free tier to try shell automation and subprocess command pipeline code generation.