plumbum wraps shell commands as Python objects. pip install plumbum. Local: from plumbum import local. ls = local["ls"]. ls() — run, return stdout. ls["-la"]() — with flag. ls["-la", "/tmp"]() — flag + arg. Bound: cmd = local["grep"]["-r", "pattern"]. cmd(".") — execute. git = local["git"]; git["commit", "-m", "msg"](). Pipeline: (ls | local["grep"]["py"])(). (cat << "input") | grep. Redirect: (ls > "/tmp/out.txt")(). (ls >> "/tmp/out.txt")() — append. (cmd > "/dev/null")() — discard. retcode: retcode, stdout, stderr = ls.run(retcode=None). cmd.run() → (rc, out, err). FG: from plumbum import FG; ls & FG — foreground, inherits terminal. BG: from plumbum import BG; future = ls & BG; future.wait(). TEE: from plumbum import TF; (ls & TF("/tmp/log.txt")). Error: from plumbum import ProcessExecutionError. cmd.run(retcode=[0,1]) — accept codes. env: with local.env(HOME="/tmp"):. cwd: with local.cwd("/tmp"):. SSH: from plumbum.machines import SshMachine; r = SshMachine("host"); r["ls"](). CLI: from plumbum import cli; class App(cli.Application):. @cli.switch("--verbose"). Claude Code generates plumbum command pipelines, shell wrappers, and SSH automation scripts.
CLAUDE.md for plumbum
## plumbum Stack
- Version: plumbum >= 1.8 | pip install plumbum
- Command: local["cmd"]["-flag", "arg"]() — run and return stdout
- Pipeline: (cmd1 | cmd2)() — pipe stdout | stdin
- Redirect: (cmd > "file")() | (cmd >> "file")() for append
- Retcode: cmd.run(retcode=None) → (rc, out, err) | retcode=[0,1] for multiple OK codes
- Env/cwd: with local.env(VAR="val"): | with local.cwd("/path"):
- Error: catch ProcessExecutionError — has .retcode .stdout .stderrplumbum Shell Pipeline
# app/shell.py — plumbum command execution, pipelines, and SSH automation
from __future__ import annotations
import sys
from pathlib import Path
from typing import Any
from plumbum import BG, FG, ProcessExecutionError, local
from plumbum.machines import LocalMachine
# ─────────────────────────────────────────────────────────────────────────────
# 1. Command wrappers
# ─────────────────────────────────────────────────────────────────────────────
def run(cmd_name: str, *args: str, retcode: int | list[int] | None = 0) -> str:
"""
Run a local command with arguments and return stdout.
retcode=None accepts any exit code.
retcode=[0,1] accepts exit codes 0 and 1.
Raises ProcessExecutionError on unexpected exit codes.
"""
cmd = local[cmd_name][args]
rc, out, err = cmd.run(retcode=retcode)
return out
def run_checked(cmd_name: str, *args: str) -> tuple[int, str, str]:
"""Run a command and return (returncode, stdout, stderr) regardless of exit code."""
cmd = local[cmd_name][args]
return cmd.run(retcode=None)
def run_lines(cmd_name: str, *args: str) -> list[str]:
"""Run command and return stdout as a list of non-empty lines."""
out = run(cmd_name, *args)
return [line for line in out.splitlines() if line.strip()]
# ─────────────────────────────────────────────────────────────────────────────
# 2. Pipeline operations
# ─────────────────────────────────────────────────────────────────────────────
def grep_files(pattern: str, path: str = ".", extension: str = "py") -> list[str]:
"""
Pipe: find ... | grep pattern — returns matching lines.
Plumbum pipeline: (cmd1 | cmd2)() runs both with connected pipes.
"""
try:
find = local["find"][path, "-name", f"*.{extension}", "-type", "f"]
xargs = local["xargs"]["grep", "-l", pattern]
result = (find | xargs)()
return [l for l in result.splitlines() if l.strip()]
except ProcessExecutionError as e:
if e.retcode == 1: # grep returns 1 when no matches
return []
raise
def count_lines(path: str) -> dict[str, int]:
"""
Pipe: find | wc — count total lines across all Python files.
Demonstrates chaining three commands in a pipeline.
"""
try:
find = local["find"][path, "-name", "*.py", "-type", "f"]
xargs = local["xargs"]["wc", "-l"]
out = (find | xargs)()
# Parse "total" line from wc -l output
for line in reversed(out.splitlines()):
parts = line.split()
if parts and parts[-1] == "total":
return {"total_lines": int(parts[0])}
return {"total_lines": 0}
except ProcessExecutionError:
return {"total_lines": 0}
def redirect_to_file(cmd_name: str, *args: str, output: str) -> None:
"""
Redirect command output to a file.
(cmd > "path")() overwrites; (cmd >> "path")() appends.
"""
cmd = local[cmd_name][args]
(cmd > output)()
# ─────────────────────────────────────────────────────────────────────────────
# 3. Git helpers
# ─────────────────────────────────────────────────────────────────────────────
_git = local["git"]
def git_status() -> str:
"""Return git status output."""
rc, out, err = _git["status", "--short"].run(retcode=None)
return out
def git_log(n: int = 10, format_str: str = "%h %s") -> list[str]:
"""Return last n commit messages."""
out = _git["log", f"-{n}", f"--pretty=format:{format_str}"]()
return [l for l in out.splitlines() if l]
def git_branch() -> str:
"""Return current branch name."""
return _git["rev-parse", "--abbrev-ref", "HEAD"]().strip()
def git_changed_files() -> list[str]:
"""Return list of modified (uncommitted) files."""
rc, out, _ = _git["diff", "--name-only"].run(retcode=None)
return [l for l in out.splitlines() if l.strip()]
def git_tag(tag_name: str, message: str = "") -> None:
"""Create an annotated tag."""
if message:
_git["tag", "-a", tag_name, "-m", message]()
else:
_git["tag", tag_name]()
# ─────────────────────────────────────────────────────────────────────────────
# 4. Environment and working directory context managers
# ─────────────────────────────────────────────────────────────────────────────
def run_in_dir(directory: str | Path, cmd_name: str, *args: str) -> str:
"""
Run a command in a specific working directory.
local.cwd() is a context manager — restores cwd on exit.
"""
with local.cwd(str(directory)):
return run(cmd_name, *args)
def run_with_env(env_vars: dict[str, str], cmd_name: str, *args: str) -> str:
"""
Run a command with additional environment variables.
local.env() adds/overrides env vars in the subprocess.
"""
with local.env(**env_vars):
return run(cmd_name, *args)
# ─────────────────────────────────────────────────────────────────────────────
# 5. Background processes
# ─────────────────────────────────────────────────────────────────────────────
def run_background(cmd_name: str, *args: str):
"""
Start a command in the background.
Returns a Future — call .wait() to block until done.
cmd & BG starts the process but doesn't wait.
"""
cmd = local[cmd_name][args]
return cmd & BG
def run_parallel(commands: list[tuple[str, ...]]) -> list[str]:
"""
Run multiple commands in parallel using BG futures.
Collects all results after all complete.
"""
futures = []
for cmd_spec in commands:
cmd_name, *args = cmd_spec
cmd = local[cmd_name][args]
futures.append(cmd & BG)
results = []
for future in futures:
future.wait()
results.append(future.stdout)
return results
# ─────────────────────────────────────────────────────────────────────────────
# 6. Process execution error handling
# ─────────────────────────────────────────────────────────────────────────────
def safe_run(cmd_name: str, *args: str, default: str = "") -> str:
"""Run a command, returning default string on failure."""
try:
return run(cmd_name, *args)
except ProcessExecutionError:
return default
except FileNotFoundError:
return default
def run_or_raise(cmd_name: str, *args: str, msg: str = "") -> str:
"""Run a command, converting ProcessExecutionError to RuntimeError with context."""
try:
return run(cmd_name, *args)
except ProcessExecutionError as e:
raise RuntimeError(
f"{msg or cmd_name} failed (exit {e.retcode}):\n{e.stderr.strip()}"
) from e
# ─────────────────────────────────────────────────────────────────────────────
# 7. SSH remote execution (optional — requires paramiko)
# ─────────────────────────────────────────────────────────────────────────────
def run_remote(host: str, cmd_name: str, *args: str, user: str = "", key: str = "") -> str:
"""
Run a command on a remote host via SSH.
SshMachine wraps paramiko for transparent remote execution.
"""
from plumbum.machines import SshMachine
kwargs: dict[str, Any] = {}
if user:
kwargs["user"] = user
if key:
kwargs["keyfile"] = key
with SshMachine(host, **kwargs) as remote:
cmd = remote[cmd_name][args]
return cmd()
# ─────────────────────────────────────────────────────────────────────────────
# Demo
# ─────────────────────────────────────────────────────────────────────────────
if __name__ == "__main__":
print("=== Basic command ===")
try:
out = run("uname", "-s")
print(f" OS: {out.strip()}")
except FileNotFoundError:
print(" uname not available")
print("\n=== Environment ===")
try:
with local.env(MY_VAR="hello_plumbum"):
out = run("printenv", "MY_VAR")
print(f" MY_VAR={out.strip()}")
except Exception as e:
print(f" {e}")
print("\n=== Git status ===")
try:
status = git_status()
lines = [l for l in status.splitlines() if l.strip()]
print(f" {len(lines)} changed file(s)")
except Exception as e:
print(f" {e}")
print("\n=== Git log (last 3) ===")
try:
for commit in git_log(n=3):
print(f" {commit}")
except Exception as e:
print(f" {e}")
print("\n=== safe_run (nonexistent command) ===")
result = safe_run("nonexistent_binary_12345", default="(not found)")
print(f" result: {result!r}")For the subprocess alternative — subprocess.run(["cmd", "arg"], capture_output=True, text=True) is the stdlib approach but requires building argument lists manually, no pipeline | operator, and verbose boilerplate for chaining; plumbum’s (cmd1 | cmd2)() syntax maps directly to shell pipe semantics, local["cmd"]["-flag"] reads like the original command, and with local.cwd("/path"): / with local.env(VAR="val"): replace shell cd and export without subprocess-level environment dicts. For the sh (python-sh) alternative — python-sh dynamically creates command wrappers as module attributes (from sh import git; git.log()) and uses more magic, while plumbum uses explicit local["git"]["log"]() lookups that are grep-able and IDE-friendly, and plumbum’s SshMachine provides the same API for remote execution — remote["ls"]() — that local["ls"]() provides locally, making it easy to write scripts that optionally run on a remote host. The Claude Skills 360 bundle includes plumbum skill sets covering local[“cmd”]args execution, run() and run_checked() helpers, pipeline operator (cmd1 | cmd2)(), redirection (cmd > file)() and (cmd >> file)(), ProcessExecutionError with retcode/stdout/stderr, run_in_dir with local.cwd() context manager, run_with_env with local.env() context manager, BG background futures, run_parallel multi-process launcher, git_status/log/branch/changed_files helpers, SshMachine remote execution, and safe_run fallback wrapper. Start with the free tier to try shell automation pipeline code generation.