TL;DR

维度 subprocess sh
类型 标准库 第三方库
语法风格 列表 / 字符串 方法链
学习曲线 中等 低
代码量 多 少
跨平台 ✅ Windows/Linux/macOS ❌ 仅 POSIX
类型提示 ✅ 完整 ⚠️ 有限
推荐场景 跨平台、生产代码 Linux 脚本、快速开发

1. 基础介绍

subprocess（标准库）

Python 3.5+ 内置的进程管理模块，是 os.system() 和 os.popen() 的现代替代品。

1
2
import subprocess
from subprocess import run, check_call, check_output, Popen, PIPE, CalledProcessError

sh（第三方库）

一个让 shell 命令调用更 Pythonic 的库，通过动态属性访问实现命令调用。

1
2
3
pip install sh
# 或
uv add sh
1
import sh

2. 基础用法对比

2.1 执行简单命令

1
2
3
4
5
6
7
8
9
10
11
12
# ===== subprocess =====
import subprocess

# 方式 1: run()（推荐）
result = subprocess.run(["ls", "-la"], capture_output=True, text=True)
print(result.stdout)

# 方式 2: check_call()（只关心是否成功）
subprocess.check_call(["echo", "hello"])

# 方式 3: check_output()（获取输出）
output = subprocess.check_output(["date"], text=True)
1
2
3
4
5
6
7
8
9
10
11
# ===== sh =====
import sh

# 直接调用，返回输出
print(sh.ls("-la"))

# 简单执行
sh.echo("hello")

# 获取输出
output = str(sh.date())

对比：sh 的语法更简洁，不需要列表包装参数。

2.2 传递参数

1
2
3
4
5
6
# ===== subprocess =====
# 必须使用列表形式（推荐，更安全）
subprocess.run(["git", "commit", "-m", "fix: bug"])

# 或者使用 shell=True（不推荐，有注入风险）
subprocess.run("git commit -m 'fix: bug'", shell=True)
1
2
3
4
5
6
7
8
9
10
# ===== sh =====
# 位置参数
sh.git.commit("-m", "fix: bug")

# 关键字参数（自动转换为 --key=value）
sh.git.commit(m="fix: bug")

# 混合使用
sh.docker.run("nginx", d=True, p="8080:80", name="web")
# 等价于: docker run -d -p 8080:80 --name web nginx

对比：sh 支持关键字参数自动转换，更接近 Python 风格。

2.3 获取命令输出

1
2
3
4
5
6
7
8
9
# ===== subprocess =====
# 获取 stdout
result = subprocess.run(["kubectl", "get", "pods"], capture_output=True, text=True)
stdout = result.stdout
stderr = result.stderr
return_code = result.returncode

# 只获取 stdout（失败时抛异常）
output = subprocess.check_output(["kubectl", "get", "pods"], text=True)
1
2
3
4
5
6
7
8
9
10
11
# ===== sh =====
# 直接返回输出（类似 check_output）
output = sh.kubectl.get.pods()  # 返回 sh.RunningCommand 对象

# 转换为字符串
stdout = str(output)

# 获取详细信息
stdout = output.stdout
stderr = output.stderr
exit_code = output.exit_code

2.4 工作目录和环境变量

1
2
3
4
5
6
7
8
# ===== subprocess =====
import os

subprocess.run(
    ["npm", "run", "build"],
    cwd="/path/to/project",
    env={**os.environ, "NODE_ENV": "production"}
)
1
2
3
4
5
# ===== sh =====
sh.npm.run.build(
    _cwd="/path/to/project",
    _env={**os.environ, "NODE_ENV": "production"}
)

注意：sh 使用下划线前缀 _cwd, _env 来区分命令参数和控制参数。

3. 进阶用法对比

3.1 管道（Pipe）

1
2
3
4
5
6
7
8
9
10
# ===== subprocess =====
from subprocess import Popen, PIPE

# cat file.txt | grep "error" | wc -l
p1 = Popen(["cat", "file.txt"], stdout=PIPE)
p2 = Popen(["grep", "error"], stdin=p1.stdout, stdout=PIPE)
p3 = Popen(["wc", "-l"], stdin=p2.stdout, stdout=PIPE, text=True)
p1.stdout.close()
p2.stdout.close()
output = p3.communicate()[0]
1
2
3
4
5
6
# ===== sh =====
# 方式 1: 嵌套调用
output = sh.wc(sh.grep(sh.cat("file.txt"), "error"), "-l")

# 方式 2: 管道操作符
output = sh.cat("file.txt") | sh.grep("error") | sh.wc("-l")

对比：sh 的管道语法极其简洁，几乎与 shell 一致。

3.2 后台执行

1
2
3
4
5
6
7
8
9
10
11
12
# ===== subprocess =====
import subprocess

# 启动后台进程
process = subprocess.Popen(["python", "server.py"])

# 继续执行其他代码...
print("Server started in background")

# 稍后等待或终止
process.terminate()
process.wait()
1
2
3
4
5
6
7
8
9
10
11
12
# ===== sh =====
# 使用 _bg=True
process = sh.python("server.py", _bg=True)

# 继续执行其他代码...
print("Server started in background")

# 等待完成
process.wait()

# 或者终止
process.terminate()

3.3 实时输出流（Streaming）

1
2
3
4
5
6
7
8
9
10
# ===== subprocess =====
process = subprocess.Popen(
    ["docker", "build", "."],
    stdout=subprocess.PIPE,
    stderr=subprocess.STDOUT,
    text=True
)
for line in process.stdout:
    print(line, end="")
process.wait()
1
2
3
4
5
6
7
8
9
10
11
12
13
# ===== sh =====
# 方式 1: 迭代器
for line in sh.docker.build(".", _iter=True):
    print(line, end="")

# 方式 2: 回调函数
def process_line(line):
    print(f"[BUILD] {line}", end="")

sh.docker.build(".", _out=process_line)

# 方式 3: 同时处理 stdout 和 stderr
sh.docker.build(".", _out=process_line, _err=process_line)

3.4 超时控制

1
2
3
4
5
6
7
8
9
# ===== subprocess =====
try:
    result = subprocess.run(
        ["long_running_command"],
        timeout=30,  # 秒
        capture_output=True
    )
except subprocess.TimeoutExpired:
    print("Command timed out")
1
2
3
4
5
# ===== sh =====
try:
    result = sh.long_running_command(_timeout=30)
except sh.TimeoutException:
    print("Command timed out")

3.5 错误处理

1
2
3
4
5
6
7
8
9
10
# ===== subprocess =====
from subprocess import CalledProcessError

try:
    subprocess.check_call(["false"])  # 返回码 1
except CalledProcessError as e:
    print(f"Command failed with code {e.returncode}")
    print(f"Command: {e.cmd}")
    print(f"Output: {e.output}")
    print(f"Stderr: {e.stderr}")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
# ===== sh =====
try:
    sh.false()  # 返回码 1
except sh.ErrorReturnCode as e:
    print(f"Command failed with code {e.exit_code}")
    print(f"Full command: {e.full_cmd}")
    print(f"Stdout: {e.stdout}")
    print(f"Stderr: {e.stderr}")

# 捕获特定错误码
except sh.ErrorReturnCode_1:
    print("Exit code 1")
except sh.ErrorReturnCode_2:
    print("Exit code 2")

# 忽略特定错误码
sh.grep("pattern", "file.txt", _ok_code=[0, 1])  # grep 没找到时返回 1

3.6 交互式命令

1
2
3
4
5
6
7
8
# ===== subprocess =====
process = subprocess.Popen(
    ["python"],
    stdin=subprocess.PIPE,
    stdout=subprocess.PIPE,
    text=True
)
stdout, stderr = process.communicate(input="print('hello')\n")
1
2
3
4
5
6
7
8
9
# ===== sh =====
# 使用 _in 参数
output = sh.python(_in="print('hello')\n")

# 或者使用 _stdin
output = sh.python(_stdin="print('hello')\n")

# 交互式输入（模拟用户输入）
sh.sudo.command(_in="password\n")

4. 高级特性

4.1 sh 的特殊命令名处理

1
2
3
4
5
6
# 命令名是 Python 关键字或包含特殊字符
sh.Command("apt-get").install("vim")  # apt-get 有连字符
sh.Command("2to3")("script.py")       # 数字开头

# 或者使用下划线替代连字符
sh.apt_get.install("vim")  # 自动转换为 apt-get

4.2 sh 的命令缓存

1
2
3
4
5
6
7
# 预先创建命令对象
docker = sh.docker
kubectl = sh.kubectl.bake("--namespace", "production")  # 预设参数

# 使用
docker.ps()
kubectl.get.pods()  # 自动带上 --namespace production

4.3 sh 的全局配置

1
2
3
4
5
6
7
8
9
import sh

# 方式 1: 修改默认参数
sh2 = sh.bake(_cwd="/tmp", _env={"LC_ALL": "C"})
sh2.ls()

# 方式 2: 创建子模块
docker = sh.docker.bake("--log-level", "error")
docker.build(".")  # 自动带上 --log-level error

5. 实际场景对比

场景 1: Docker 镜像构建

1
2
3
4
5
6
7
8
9
10
11
12
13
# ===== subprocess =====
import os
from subprocess import check_call

cmd = [
    "docker", "buildx", "build",
    "--secret", "id=token,env=GITHUB_TOKEN",
    "--build-arg", f"VERSION={version}",
    "-t", f"myapp:{tag}",
    "-f", "Dockerfile",
    "."
]
check_call(cmd, env={**os.environ, "DOCKER_BUILDKIT": "1"})
1
2
3
4
5
6
7
8
9
10
11
12
# ===== sh =====
import os
import sh

sh.docker.buildx.build(
    "--secret", "id=token,env=GITHUB_TOKEN",
    "--build-arg", f"VERSION={version}",
    "-t", f"myapp:{tag}",
    "-f", "Dockerfile",
    ".",
    _env={**os.environ, "DOCKER_BUILDKIT": "1"}
)

场景 2: Git 操作

1
2
3
4
5
6
7
8
9
10
# ===== subprocess =====
from subprocess import check_output, check_call

# 获取当前分支
branch = check_output(["git", "branch", "--show-current"], text=True).strip()

# 提交
check_call(["git", "add", "."])
check_call(["git", "commit", "-m", "feat: new feature"])
check_call(["git", "push", "origin", branch])
1
2
3
4
5
6
7
8
9
10
# ===== sh =====
import sh

# 获取当前分支
branch = str(sh.git.branch("--show-current")).strip()

# 提交
sh.git.add(".")
sh.git.commit(m="feat: new feature")
sh.git.push("origin", branch)

场景 3: Kubernetes 操作

1
2
3
4
5
6
7
8
9
10
11
12
# ===== subprocess =====
import json
from subprocess import check_output

# 获取 Pod 列表
output = check_output(
    ["kubectl", "get", "pods", "-n", "default", "-o", "json"],
    text=True
)
pods = json.loads(output)
for pod in pods["items"]:
    print(pod["metadata"]["name"])
1
2
3
4
5
6
7
8
9
10
11
12
# ===== sh =====
import json
import sh

# 获取 Pod 列表
output = sh.kubectl.get.pods("-n", "default", "-o", "json")
pods = json.loads(str(output))
for pod in pods["items"]:
    print(pod["metadata"]["name"])

# 或者使用管道
names = sh.jq(".items[].metadata.name", sh.kubectl.get.pods("-o", "json"))

场景 4: 日志监控

1
2
3
4
5
6
7
8
9
10
# ===== subprocess =====
from subprocess import Popen, PIPE

process = Popen(["tail", "-f", "/var/log/app.log"], stdout=PIPE, text=True)
try:
    for line in process.stdout:
        if "ERROR" in line:
            send_alert(line)
except KeyboardInterrupt:
    process.terminate()
1
2
3
4
5
6
7
8
9
10
11
# ===== sh =====
import sh

def on_line(line):
    if "ERROR" in line:
        send_alert(line)

try:
    sh.tail("-f", "/var/log/app.log", _out=on_line)
except KeyboardInterrupt:
    pass  # sh 自动处理进程清理

6. 性能对比

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
import time
import subprocess
import sh

# 测试 1000 次简单命令
N = 1000

# subprocess
start = time.time()
for _ in range(N):
    subprocess.run(["echo", "test"], capture_output=True)
print(f"subprocess: {time.time() - start:.2f}s")

# sh
start = time.time()
for _ in range(N):
    sh.echo("test")
print(f"sh: {time.time() - start:.2f}s")

典型结果：

  • subprocess: ~2.5s
  • sh: ~3.0s

结论：sh 略慢（约 20%），主要是动态属性查找的开销。对于大多数场景，这个差异可以忽略。

7. 迁移指南

7.1 从 subprocess 迁移到 sh

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
# Before (subprocess)
from subprocess import check_call, check_output, CalledProcessError

try:
    check_call(["git", "pull"])
    output = check_output(["git", "log", "-1"], text=True)
except CalledProcessError as e:
    print(f"Error: {e}")

# After (sh)
import sh

try:
    sh.git.pull()
    output = str(sh.git.log("-1"))
except sh.ErrorReturnCode as e:
    print(f"Error: {e}")

7.2 迁移对照表

subprocess sh
check_call(["cmd", "arg"]) sh.cmd("arg")
check_output([...], text=True) str(sh.cmd(...))
run([...], capture_output=True) sh.cmd(...)
Popen([...]) sh.cmd(..., _bg=True)
CalledProcessError sh.ErrorReturnCode
cwd="path" _cwd="path"
env={...} _env={...}
timeout=30 _timeout=30
stdin=PIPE _in="input"
stdout=PIPE 默认捕获

8. 最佳实践

何时使用 subprocess

  1. 需要跨平台支持（Windows）
  2. 生产环境代码（标准库更稳定）
  3. 需要完整的类型提示
  4. 团队不熟悉 sh

何时使用 sh

  1. Linux/macOS 专用脚本
  2. DevOps 工具和 CLI
  3. 快速原型开发
  4. 大量管道操作
  5. 需要实时输出流处理

混合使用

1
2
3
4
5
6
7
8
9
10
11
12
13
14
# 可以在同一项目中混合使用
import subprocess
import sh

# 简单命令用 sh
output = sh.git.status(s=True)

# 复杂场景用 subprocess
process = subprocess.Popen(
    ["complex", "command"],
    stdin=subprocess.PIPE,
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE
)

9. 常见陷阱

sh 的陷阱

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
# 1. 命令不存在时的错误
try:
    sh.nonexistent_command()
except sh.CommandNotFound:
    print("Command not found")

# 2. 参数中的空格
sh.echo("hello world")  # ✅ 正确
sh.echo("hello", "world")  # ✅ 也正确，两个参数

# 3. 布尔参数
sh.ls(l=True)   # ls -l
sh.ls(l=False)  # ls（不带 -l）

# 4. 长选项
sh.git.log(oneline=True)  # git log --oneline
sh.git.log(n=5)           # git log -n 5

subprocess 的陷阱

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
# 1. shell=True 的安全风险
user_input = "file.txt; rm -rf /"
subprocess.run(f"cat {user_input}", shell=True)  # 危险！

# 正确做法
subprocess.run(["cat", user_input])  # 安全

# 2. 忘记处理 stderr
result = subprocess.run(["cmd"], capture_output=True)
if result.returncode != 0:
    print(result.stderr)  # 别忘了检查 stderr

# 3. 死锁风险
process = subprocess.Popen(["cmd"], stdout=PIPE, stderr=PIPE)
# 如果输出太多，communicate() 之前会死锁
stdout, stderr = process.communicate()  # 使用 communicate() 避免死锁

10. 总结

个人迁移计划

  1. 新脚本：优先使用 sh
  2. 现有代码：逐步迁移，不急于一次性替换
  3. 生产代码：保持 subprocess，确保稳定性
  4. 跨平台项目：继续使用 subprocess

推荐工具链

1
2
3
4
5
6
7
# 开发脚本标准导入
import sh
from sh import git, docker, kubectl

# 预设常用命令
k = sh.kubectl.bake("-n", "default")
dc = sh.docker.compose.bake("-f", "docker-compose.yml")

11. 其他竞品对比

除了 subprocess 和 sh，还有几个值得了解的库：

11.1 Plumbum

“Shell combinators and more” - 功能最丰富的竞品

1
pip install plumbum
1
2
3
4
5
6
7
8
9
10
11
12
13
14
from plumbum import local
from plumbum.cmd import git, grep, wc

# 基本用法
print(git["status"]())

# 管道
chain = git["log"] | grep["fix"] | wc["-l"]
print(chain())

# 远程执行
from plumbum import SshMachine
with SshMachine("server.com") as remote:
    print(remote["ls"]("-la"))

特点：

  • ✅ 支持远程执行（SSH）
  • ✅ 支持 Windows
  • ✅ 路径操作集成
  • ✅ 颜色终端支持
  • ⚠️ 语法比 sh 复杂

11.2 Invoke

Fabric 的基础库，专注于任务自动化

1
pip install invoke
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from invoke import run, Context

# 简单执行
result = run("git status")
print(result.stdout)

# 任务定义（类似 Makefile）
from invoke import task

@task
def build(c):
    c.run("docker build -t myapp .")

@task
def deploy(c, env="staging"):
    c.run(f"kubectl apply -f k8s/{env}/")

特点：

  • ✅ 任务编排能力强
  • ✅ 支持任务依赖
  • ✅ 内置帮助生成
  • ⚠️ 主要用于任务定义，不是通用命令执行

11.3 Pexpect

专门用于交互式命令

1
pip install pexpect
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
import pexpect

# 自动化 SSH 登录
child = pexpect.spawn("ssh user@server")
child.expect("password:")
child.sendline("mypassword")
child.expect("$")
child.sendline("ls -la")
child.expect("$")
print(child.before.decode())

# 自动化 sudo
child = pexpect.spawn("sudo apt update")
child.expect("[sudo] password")
child.sendline("password")
child.expect(pexpect.EOF)

特点：

  • ✅ 交互式命令的最佳选择
  • ✅ 模式匹配（expect）
  • ✅ 超时控制
  • ⚠️ 不适合简单命令
  • ❌ 不支持 Windows（用 wexpect 替代）

11.4 Delegator.py

Kenneth Reitz（requests 作者）的作品，极简主义

1
pip install delegator.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
import delegator

# 简单执行
c = delegator.run("ls -la")
print(c.out)
print(c.return_code)

# 管道
c = delegator.chain("cat file.txt | grep error | wc -l")
print(c.out)

# 后台执行
c = delegator.run("sleep 10", block=False)
c.kill()

特点：

  • ✅ API 极简
  • ✅ 管道语法直观
  • ⚠️ 功能较少
  • ⚠️ 维护不活跃

11.5 对比总表

库 语法风格 Windows 远程执行 交互式 活跃度 Stars
subprocess 标准库 ✅ ❌ ⚠️ ✅ N/A
sh 方法链 ❌ ❌ ⚠️ ✅ 6.9k
plumbum 索引式 ✅ ✅ SSH ⚠️ ✅ 2.8k
invoke 装饰器 ✅ ✅ Fabric ⚠️ ✅ 4.3k
pexpect 交互式 ❌ ⚠️ ✅ ✅ 2.6k
delegator 极简 ⚠️ ❌ ❌ ⚠️ 1.7k

11.6 选择建议 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
 需要跨平台？
├── 是 → subprocess 或 plumbum
└── 否 → 继续 ↓

需要远程执行？
├── 是 → plumbum 或 invoke + fabric
└── 否 → 继续 ↓

需要交互式？
├── 是 → pexpect
└── 否 → 继续 ↓

需要任务编排？
├── 是 → invoke
└── 否 → 继续 ↓

追求简洁语法？
├── 是 → sh ⭐（推荐）
└── 否 → subprocess

11.7 快速示例对比

同一个任务：执行 git log -5 --oneline

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
# subprocess
from subprocess import check_output
output = check_output(["git", "log", "-5", "--oneline"], text=True)

# sh
import sh
output = str(sh.git.log("-5", "--oneline"))

# plumbum
from plumbum.cmd import git
output = git["log", "-5", "--oneline"]()

# invoke
from invoke import run
result = run("git log -5 --oneline", hide=True)
output = result.stdout

# delegator
import delegator
output = delegator.run("git log -5 --oneline").out

参考资料