sh turns shell commands into Python functions. pip install sh. Run: from sh import ls; ls("-la"). Import: from sh import git, grep, ls, cat, echo. Output: output = ls("-la"); str(output). Args: git("log", "--oneline", "-n", "5"). Keyword args: git.log(oneline=True, n=5) — --oneline -n 5. Chaining: grep(ls("/etc"), "conf"). Pipe: grep(cat("/etc/hosts"), "localhost"). stdin: wc(echo("hello world"), l=True). _in: sh.wc(_in="line1\nline2", l=True). _out: sh.ls(_out="list.txt") — redirect to file. _err: sh.git("status", _err="/dev/null"). _cwd: git("status", _cwd="/path/to/repo"). _env: sh.python("script.py", _env={"PATH": "/usr/bin"}). _timeout: sh.sleep("10", _timeout=2) → raises TimeoutExpired. _bg: p = sh.sleep("30", _bg=True); p.wait(); p.kill(). _iter: for line in sh.tail("-f", "log.txt", _iter=True): .... ErrorReturnCode: try: sh.ls("/noexist") except sh.ErrorReturnCode as e: e.exit_code. Special: from sh import Command; custom = Command("/path/custom"). sudo: from sh.contrib import sudo; sudo.apt("install", "vim"). bash: sh.bash("-c", "echo $HOME"). Glob: sh.ls(sh.glob("/tmp/*.log")). Environment: sh.env(). Claude Code generates sh automation scripts, git wrappers, build pipelines, and process managers.
CLAUDE.md for sh
## sh Stack
- Version: sh >= 2.0 | pip install sh | Unix/macOS only (not Windows)
- Import: from sh import git, ls, grep, find, cat | Command("/bin/custom")
- Args: cmd("--flag", "value") | cmd(flag=True, n=5) → --flag -n 5
- Control: _cwd="/path" | _env={"KEY":"VAL"} | _timeout=30 | _bg=True
- Piping: grep(ls("/dir"), "pattern") | wc(cat("file"), l=True)
- Errors: except sh.ErrorReturnCode as e: e.exit_code | e.stdout | e.stderrsh Shell Automation Pipeline
# app/shell.py — sh commands, git wrapper, build runner, file ops, process control
from __future__ import annotations
import logging
import os
import sys
from contextlib import contextmanager
from pathlib import Path
from typing import Any, Generator, Iterator
import sh
from sh import Command, ErrorReturnCode
log = logging.getLogger(__name__)
# ─────────────────────────────────────────────────────────────────────────────
# 1. Basic command helpers
# ─────────────────────────────────────────────────────────────────────────────
def run(
cmd: str,
*args: str,
cwd: str | Path | None = None,
env: dict | None = None,
timeout: int | None = None,
ok_codes: tuple[int, ...] = (0,),
) -> str:
"""
Run a shell command and return stdout as a string.
Raises sh.ErrorReturnCode on non-zero exit (unless in ok_codes).
Example:
output = run("git", "log", "--oneline", "-n", "10", cwd="/repo")
version = run("python", "--version")
"""
command = Command(cmd)
kwargs: dict[str, Any] = {}
if cwd:
kwargs["_cwd"] = str(cwd)
if env:
kwargs["_env"] = env
if timeout:
kwargs["_timeout"] = timeout
if ok_codes != (0,):
kwargs["_ok_code"] = list(ok_codes)
result = command(*args, **kwargs)
return str(result).strip()
def run_lines(
cmd: str,
*args: str,
cwd: str | Path | None = None,
) -> list[str]:
"""
Run a command and return output as list of non-empty lines.
Example:
files = run_lines("ls", "-1", "/tmp")
pids = run_lines("pgrep", "python")
"""
output = run(cmd, *args, cwd=cwd)
return [line for line in output.splitlines() if line.strip()]
def run_silent(cmd: str, *args: str, **kwargs) -> bool:
"""
Run a command; return True on success, False on non-zero exit.
Example:
if run_silent("which", "ffmpeg"):
print("ffmpeg is installed")
"""
try:
Command(cmd)(*args, _out=os.devnull, _err=os.devnull, **kwargs)
return True
except ErrorReturnCode:
return False
@contextmanager
def in_dir(path: str | Path) -> Generator[None, None, None]:
"""
Context manager: temporarily change working directory.
Example:
with in_dir("/repo"):
output = run("git", "status")
"""
original = os.getcwd()
try:
os.chdir(str(path))
yield
finally:
os.chdir(original)
# ─────────────────────────────────────────────────────────────────────────────
# 2. Streaming and background processes
# ─────────────────────────────────────────────────────────────────────────────
def stream_lines(
cmd: str,
*args: str,
cwd: str | Path | None = None,
timeout: int | None = None,
) -> Iterator[str]:
"""
Stream command output line-by-line (for long-running commands).
Example:
for line in stream_lines("tail", "-f", "/var/log/app.log"):
if "ERROR" in line:
alert(line)
for line in stream_lines("make", "build"):
print(line, end="")
"""
command = Command(cmd)
kwargs: dict[str, Any] = {"_iter": True}
if cwd:
kwargs["_cwd"] = str(cwd)
if timeout:
kwargs["_timeout"] = timeout
for line in command(*args, **kwargs):
yield str(line).rstrip("\n")
def run_background(
cmd: str,
*args: str,
cwd: str | Path | None = None,
env: dict | None = None,
) -> sh.RunningCommand:
"""
Run a command in the background (non-blocking).
Call .wait() to block until done, .kill() to terminate.
Example:
server = run_background("python", "-m", "http.server", "8000")
time.sleep(1)
requests.get("http://localhost:8000")
server.kill()
"""
command = Command(cmd)
kwargs: dict[str, Any] = {"_bg": True}
if cwd:
kwargs["_cwd"] = str(cwd)
if env:
kwargs["_env"] = env
return command(*args, **kwargs)
# ─────────────────────────────────────────────────────────────────────────────
# 3. Git wrapper
# ─────────────────────────────────────────────────────────────────────────────
class GitRepo:
"""
High-level sh-backed git operations for a specific repo directory.
Example:
repo = GitRepo("/path/to/repo")
print(repo.current_branch())
repo.pull()
repo.commit("Auto-commit: update configs")
"""
def __init__(self, path: str | Path = ".") -> None:
self.path = str(path)
self._git = lambda *args, **kw: Command("git")(*args, _cwd=self.path, **kw)
def status(self, short: bool = True) -> str:
args = ["status", "--short"] if short else ["status"]
return str(self._git(*args)).strip()
def current_branch(self) -> str:
return str(self._git("rev-parse", "--abbrev-ref", "HEAD")).strip()
def log(self, n: int = 10, oneline: bool = True) -> list[str]:
args = ["log", f"-{n}"]
if oneline:
args.append("--oneline")
return [l for l in str(self._git(*args)).strip().splitlines() if l]
def diff(self, staged: bool = False) -> str:
args = ["diff"]
if staged:
args.append("--staged")
return str(self._git(*args)).strip()
def add(self, *files: str) -> None:
self._git("add", *files)
def commit(self, message: str, allow_empty: bool = False) -> str:
args = ["commit", "-m", message]
if allow_empty:
args.append("--allow-empty")
return str(self._git(*args)).strip()
def push(self, remote: str = "origin", branch: str | None = None) -> str:
args = ["push", remote]
if branch:
args.append(branch)
return str(self._git(*args)).strip()
def pull(self, remote: str = "origin", branch: str | None = None) -> str:
args = ["pull", remote]
if branch:
args.append(branch)
return str(self._git(*args)).strip()
def checkout(self, branch: str, create: bool = False) -> None:
args = ["checkout"]
if create:
args.append("-b")
args.append(branch)
self._git(*args)
def ls_files(self) -> list[str]:
return [l for l in str(self._git("ls-files")).strip().splitlines() if l]
def is_dirty(self) -> bool:
return bool(self.status().strip())
def last_commit_hash(self, short: bool = True) -> str:
args = ["rev-parse"]
if short:
args.append("--short")
args.append("HEAD")
return str(self._git(*args)).strip()
# ─────────────────────────────────────────────────────────────────────────────
# 4. File and system utilities
# ─────────────────────────────────────────────────────────────────────────────
def find_files(
directory: str | Path,
pattern: str = "*",
type_: str = "f",
max_depth: int | None = None,
) -> list[str]:
"""
Find files matching a pattern using the `find` command.
Example:
py_files = find_files("src", "*.py")
dirs = find_files(".", type_="d", max_depth=2)
"""
args = [str(directory), "-name", pattern, "-type", type_]
if max_depth is not None:
args += ["-maxdepth", str(max_depth)]
from sh import find
return [l for l in str(find(*args)).strip().splitlines() if l]
def grep_files(
pattern: str,
path: str | Path = ".",
recursive: bool = True,
ignore_case: bool = False,
file_pattern: str | None = None,
) -> list[str]:
"""
Grep for a pattern in files.
Example:
hits = grep_files("TODO", "src", recursive=True, file_pattern="*.py")
"""
from sh import grep
args = [pattern]
if recursive:
args.append("-r")
if ignore_case:
args.append("-i")
if file_pattern:
args += ["--include", file_pattern]
args.append(str(path))
try:
return [l for l in str(grep(*args)).strip().splitlines() if l]
except ErrorReturnCode as e:
if e.exit_code == 1: # no matches
return []
raise
def disk_usage(path: str | Path = ".") -> str:
"""Return human-readable disk usage for a path."""
from sh import du
return str(du("-sh", str(path))).strip().split("\t")[0]
def which(cmd: str) -> str | None:
"""Return path to command or None if not found."""
try:
from sh import which as _which
return str(_which(cmd)).strip()
except ErrorReturnCode:
return None
# ─────────────────────────────────────────────────────────────────────────────
# 5. Error handling
# ─────────────────────────────────────────────────────────────────────────────
def safe_run(cmd: str, *args, default: str = "", **kwargs) -> str:
"""
Run a command and return default on failure instead of raising.
Example:
version = safe_run("git", "--version", default="git not found")
"""
try:
return run(cmd, *args, **kwargs)
except (ErrorReturnCode, sh.CommandNotFound) as e:
log.debug("Command failed: %s %s — %s", cmd, args, e)
return default
# ─────────────────────────────────────────────────────────────────────────────
# Demo
# ─────────────────────────────────────────────────────────────────────────────
if __name__ == "__main__":
print("=== Basic run ===")
print(f" ls /tmp (first 3): {run_lines('ls', '/tmp')[:3]}")
print(f" which python: {which('python3') or 'not found'}")
print(f" disk_usage /tmp: {disk_usage('/tmp')}")
print("\n=== run_silent ===")
print(f" ffmpeg installed: {run_silent('which', 'ffmpeg')}")
print(f" python installed: {run_silent('which', 'python3')}")
print("\n=== safe_run ===")
v = safe_run("git", "--version", default="git not found")
print(f" git --version: {v}")
print("\n=== grep_files ===")
py_files = find_files(".", "*.py", max_depth=2)
print(f" *.py files (cwd, depth 2): {py_files[:3]}")
print("\n=== GitRepo ===")
try:
repo = GitRepo(".")
print(f" branch: {repo.current_branch()}")
print(f" is dirty: {repo.is_dirty()}")
for entry in repo.log(n=3):
print(f" {entry}")
except Exception as e:
print(f" (not a git repo: {e})")
print("\n=== stream_lines (echo) ===")
for line in stream_lines("echo", "line1\nline2\nline3"):
print(f" streamed: {line!r}")
print("\nInstall: pip install sh (Unix/macOS only — use subprocess on Windows)")For the subprocess alternative — Python’s built-in subprocess.run() / subprocess.Popen() works on all platforms including Windows and provides full control over process lifecycle; sh wraps subprocess with a much cleaner API where any Unix command becomes a Python function — use subprocess for portable production code and Windows support, sh for Unix-only automation scripts and DevOps tooling where the cleaner syntax reduces boilerplate significantly. For the fabric alternative — Fabric builds on Paramiko for SSH-based remote execution (deploy to servers, run commands over SSH, manage fleets); sh is purely for local subprocess execution — use Fabric when you need to run commands on remote hosts, sh when you’re orchestrating local shell commands from Python build scripts, test runners, or CLI tools. The Claude Skills 360 bundle includes sh skill sets covering run()/run_lines()/run_silent() execution, in_dir() context manager, stream_lines() iterator, run_background() process control, GitRepo wrapper with status/log/commit/push/pull/checkout, find_files()/grep_files()/disk_usage()/which() utilities, safe_run() error-silent excecution. Start with the free tier to try shell command automation and subprocess scripting code generation.