Python sh库完整翻译

发表于 更新于 分类于

什么是 sh

sh 是一个成熟的 subprocess 替代品，适用于 Python2 - 3，PyPy 和 PyPy3，它允许你以函数的方式调用任何 Shell 程序。就像这样：

1
2
from sh import ifconfig
print(ifconfig("wlan"))

输出为：

1
2
3
4
5
6
7
8
wlan0   Link encap:Ethernet  HWaddr 00:00:00:00:00:00
        inet addr:192.168.1.100  Bcast:192.168.1.255  Mask:255.255.255.0
        inet6 addr: ffff::ffff:ffff:ffff:fff/64 Scope:Link
        UP BROADCAST RUNNING MULTICAST  MTU:1500  Metric:1
        RX packets:0 errors:0 dropped:0 overruns:0 frame:0
        TX packets:0 errors:0 dropped:0 overruns:0 carrier:0
        collisions:0 txqueuelen:1000
        RX bytes:0 (0 GB)  TX bytes:0 (0 GB)

注意了，这可不是 Python 的功能，这些都是通过解析$PATH变量运行的你的系统上的二进制命令，就和 Bash 一样，然后转换为 Python 中的函数。这样，系统中的所有程序都可以轻松的用 Python 调用了。

sh 模块依赖 Unix 系统调用，所以只能在 Unix-like 操作系统上工作——比如 Linux，macOS，BSD，等等，也就是说，不支持 Windows。

注：由于 Python 不允许函数名中包含-，因此包含-的命令中的-会被转换为_。

安装

1
pip install sh

快速参考

传参

1
sh.ls("-l", "/tmp", color="never")

阅读更多

退出状态码

1
2
3
4
try:
    sh.ls("/doesnt/exist")
except sh.ErrorReturnCode_2:
    print("directory doesn't exist")

阅读更多

输出重定向

1
2
3
4
5
6
7
8
sh.ls(_out="/tmp/dir_contents")

with open("/tmp/dir_contents", "w") as h:
    sh.ls(_out=h)

from io import StringIO
buf = StringIO()
sh.ls(_out=buf)

阅读更多

参数烘焙

1
2
3
4
5
my_ls = sh.ls.bake("-l")

# 等价
my_ls("/tmp")
sh.ls("-l", "/tmp")

阅读更多

管道

1
sh.wc(_in=sh.ls("-1"), "-l")

阅读更多

子命令

1
2
3
# 等价
sh.git("show", "HEAD")
sh.git.show("HEAD")

阅读更多

异步进程

1
2
3
p = sh.find("-name", "sh.py", _bg=True)
# ... 随便写点别的 ...
p.wait()

阅读更多

传参详解

多重参数

给一条命令传递多个参数时，每个参数必须是用逗号分离的字符串：

1
2
from sh import tar
tar("cvf", "/tmp/test.tar", "/my/home/directory/")

以下这种方式是错误的：

1
2
from sh import tar
tar("cvf /tmp/test.tar /my/home/directory")

关键字参数

sh 不仅完全支持短型命令选项-a和长型命令选项--arg，而且还支持把命令选项当作关键字来传参：

1
2
3
4
5
6
7
8
9
10
11
12
# 解释为 "curl http://duckduckgo.com/ -o page.html --silent"
curl("http://duckduckgo.com/", o="page.html", silent=True)

# 或者如果你不想用关键字传参，这样也一样：
curl("http://duckduckgo.com/", "-o", "page.html", "--silent")

# 解释为 "adduser amoffat --system --shell=/bin/bash --no-create-home"
# 不要忘了 Python 要求必须把位置参数前置
adduser("amoffat", system=True, shell="/bin/bash", no_create_home=True)

# 或者
adduser("amoffat", "--system", "--shell", "/bin/bash", "--no-create-home")

退出码与异常

众所周知，正常进程结束后会以 0 退出码退出。要获取退出码，我们首先需要将 sh 模块的函数的返回值修改为RunningCommand对象，然后从对象中提取出RunningCommand.exit_code属性：

1
2
3
4
# 设置返回命令对象本身
output = ls("/", _return_cmd=True)
# 获取退出码
print(output.exit_code)

如果一个进程中止了，那么退出码就不会是 0，一个异常也会自动抛出，这种情况下你是无法通过RunningCommand.exit_code获取退出码的。不过，你可以通过ErrorReturnCode_X捕获这些特定的退出码，也可以干脆通过错误基类ErrorReturnCode把它们全部捕获：

1
2
3
4
5
6
7
try:
    print(sh.ls("/some/non-existant/folder"))
except sh.ErrorReturnCode_2:
    print("Folder doesn't exist!")
    create_the_folder()
except sh.ErrorReturnCode:
    print("Unknown error!")

提示
可以通过_ok_code=[]参数自定义代表正常退出的退出码，默认只有 0。也就是说你可以这样做：

1
2
if sh.ls("/some/non-existant/folder", _return_cmd=True, _ok_code=[0, 2]).exit_code != 0:
    print("Unknown error!")

信号

信号会在进程因接收到信号而终止时被一并抛出。此时抛出的异常类型为SignalException，它是ErrorReturnCode的子类：

1
2
3
4
5
try:
    p = sh.sleep(3, _bg=True)
    p.kill()
except sh.SignalException_SIGKILL:
    print("killed")

提示
通过数字或信号名捕获SignalException的效果是一样的。比如，以下两种异常类是等价的：
assert sh.SignalException_SIGKILL == sh.SignalException_9

提示
使用_timeout=参数给命令限时，使用_timeout_signal参数给命令指定超时后发送的信号。

重定向详解

默认情况下，sh 将命令的标准输出字符串作为函数的返回值返回，不过 sh 也可以重定向进程的标准输出和标准错误输出到各种不同类型的目标，只需要使用_out和_err这两个特殊的动态参数。

重定向到文件名

如果指定了一串字符串，它会被假定为是文件名。文件会以"wb"权限打开，这意味着覆写模式和二进制模式。

1
2
import sh
sh.ifconfig(_out="/tmp/interfaces")

提示
怎么追加到文件呢？
使用open()函数创建一个打开的文件对象，然后在对象中追加即可：

1
2
3
h = open("/xxx", "a")

sh.ls(_out=h)

提示

  • 使用_no_out=True参数关闭标准输出。
  • 使用_no_err=True参数关闭错误输出。
  • 使用_err_to_out=True使得错误输出混合到标准输出中。
  • 使用_encoding=指定编码。
  • 使用_tee='err|out|True'实现三向重定向。

类文件对象

任何支持.write(data)方法的对象，都可以作为_out=参数的值，例如io.StringIO：

1
2
3
4
5
6
import sh
from io import StringIO

buf = StringIO()
sh.ifconfig(_out=buf)
print(buf.getvalue())

函数回调

回调函数也可以作为目标。函数必须符合以下三种签名的任何一种：

  • fn(data)
    这个函数会从进程中接受数据。
  • fn(data, stdin_queue)
    和前一个签名相比，这个函数还接受了一个queue.Queue参数，它可以用来和进程间进行交流。
  • fn(data, stdin_queue, process)
    和前一个签名相比，这个函数还接受了一个weakref.weakref参数给 OProc 对象。

异步执行

sh 提供了一些执行命令并用非阻塞的方式获取输出的方法。

增量迭代

你可以通过将函数搭配_iter这一特殊的动态参数实现以迭代的方式创建异步命令。这会创建一个Iterable（可迭代对象）（在某些特殊情况下，是一个生成器对象），你可以利用它进行循环：

1
2
3
4
5
from sh import tail

# 这会永远运行下去
for line in tail("-f", "/var/log/some_log_file.log", _iter=True):
    print(line)

默认情况下，_iter会根据标准输出进行迭代，但是你也可以通过给_iter传递"err"或"out"来改变（相对于True）。默认情况下，输出是逐行缓冲的，所以循环只会在输出产生换行的时候才能进行。你也可以通过修改命令的输出的缓冲区设置_out_bufsize来修改这一设定。

提示
如果你需要创建一个完全非阻塞的迭代器，用_iter_noblock。如果当前的迭代会阻塞进程，那么就会返回errno.EWOULDBLOCK的错误，反之你就会正常接收到输出流。

异步进程

默认情况下，每个运行的命令都会阻塞进程直到完成。如果有一个长时间运行的命令，可以让它以异步方式运行，只需要使用_bg=True这个特殊的动态参数：

1
2
3
4
5
6
7
8
9
# 会阻塞
sleep(3)
print("...3 seconds later")

# 不会阻塞
p = sleep(3, _bg=True)
print("prints immediately!")
p.wait()
print("...and 3 seconds later")

注意，你需要调用RunningCommand.wait()来保证命令完成后才退出程序。

后台运行的命令会无视SIGHUP信号（nohup），这意味这如果它们的控制进程（如果存在控制终端的话，就是 Session leader）退出的话，它们是不会退出的。

但是，由于 sh 模块下的命令默认情况下会创建它们自己的新会话，这也就是说它们自己就是自己的 Session leader，是否无视 SIGHUP 一般情况下并没有区别（在 sh 2.0 以后的版本中，_new_session=False是默认行为）。

会产生影响的情况是使用了_new_session=False，它会使得当前运行 Python 的 Shell 作为控制进程，退出当前 Shell 时一般会发送SIGHUP给子进程。

输出回调

与_bg=True相结合，sh 可以使用回调函数来增量处理输出，只需要给_out和/或_err传递一个可以调用的函数。这个函数会在每次命令产生数据流时调用。

1
2
3
4
5
6
7
from sh import tail

def process_output(line):
    print(line)

p = tail("-f", "/var/log/some_log_file.log", _out=process_output, _bg=True)
p.wait()

要控制回调函数是以行模式接收还是以块模式接收，使用_out_bufsize，要“退出”回调函数，只需要返回True值。这会告诉这条命令不要再调用你的回调函数了。

回调函数接收到的行或块可以是字符串或字节类型。如果输出可以被以指定的方式解码，那么就会传递字符串类型，否则就是纯字节类型。

提示
返回True并不会杀死进程，它只是阻止回调函数继续被调用。

交互式回调

通过特定的回调函数签名，命令可以与底层进程进行交互式交流。每个通过 sh 运行的命令都有一个内部的标准输入队列queue.Queue对象，它可以被回调函数使用。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
def interact(line, stdin):
    if line == "What... is the air-speed velocity of an unladen swallow?":
        stdin.put("What do you mean? An African or European swallow?")

    elif line == "Huh? I... I don't know that....AAAAGHHHHHH":
        cross_bridge()
        return True

    else:
        stdin.put("I don't know....AAGGHHHHH")
        return True

p = sh.bridgekeeper(_out=interact, _bg=True)
p.wait()

提示
如果你使用queue，你可以用None标记输入结束（EOF）。你也可以通过给你的回调函数添加第三个接收进程对象的参数来杀死或中止你的进程（或者给它们发送信号，真的没骗你）：

1
2
3
4
5
6
7
def process_output(line, stdin, process):
    print(line)
    if "ERROR" in line:
        process.kill()
        return True

p = tail("-f", "/var/log/some_log_file.log", _out=process_output, _bg=True)

以上代码运行后，会打印some_log_file.log的内容，直到单词"WORD"出现，然后 tail 进程会被杀死，脚本结束。

提示
你也可以用RunningCommand.terminate()发送 SIGTERM 信号，或者用RunningCommand.signal()发送其他任何信号。

完成回调

进程退出后会执行完成回调函数，既可以是正常调用（响应成功或错误的退出代码），也可以通过信号调用。它总是会被调用。

下面是一个使用_done创建多进程池的例子，sh.your_parallel_command会被同时执行不超过十次：

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

pool = Semaphore(10)

def done(cmd, success, exit_code):
    pool.release()

def do_thing(arg):
    pool.acquire()
    return sh.your_parallel_command(arg, _bg=True, _done=done)

procs = []
for arg in range(100):
    procs.append(do_thing(arg))

# essentially a join
[p.wait() for p in procs]

参数烘焙详解

sh 可以把一些参数“打包”进命令中。这本质上其实就是偏函数，就像用functools.partial()实现的那样：

1
2
3
4
5
6
7
from sh import ls

ls = ls.bake("-la")
print(ls)  # "/usr/bin/ls -la"

# 解释为 "ls -la /"
print(ls("/"))

这里的思路是让每次调用ls都带上-la参数。参数烘焙在你想和子命令结合时尤其有用：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
from sh import ssh

# 在服务器上调用whoami，这会要打很多字，尤其是你想在一台服务器上回调很多命令（而不只是whoami）的话
iam1 = ssh("myserver.com", "-p 1393", "whoami")

# 把常用参数打包进ssh不爽吗？
myserver = ssh.bake("myserver.com", p=1393)

print(myserver)  # "/usr/bin/ssh myserver.com -p 1393"

# 解释为 "/usr/bin/ssh myserver.com -p 1393 whoami"
iam2 = myserver.whoami()

assert(iam1 == iam2)  # 返回True！

只要myserver这个 Callable 代表着打包过的ssh命令，你就可以方便的调用在远程服务器上的任何命令了：

1
2
# 执行 "/usr/bin/ssh myserver.com -p 1393 tail /var/log/dumb_daemon.log -n 100"
print(myserver.tail("/var/log/dumb_daemon.log", n=100))

你还可以烘焙 sh 模块本身，以配置全局适用的参数：

1
2
3
# 关闭终端友好的输出
mysh = sh.bake(_tty_out=False)
mysh.ls()

管道详解

基础

Bash 风格的管道可以使用函数嵌套的方式实现。只需要把一个命令作为给另一个命令的_in参数即可，sh 就会自动把内部命令的输出传给外部命令的输入：

1
2
3
4
5
# 按大小顺序给目录文件分类
print(sort(_in=du(glob("*"), "-sb"), "-rn"))

# 打印 /etc 中的目录和文件数量
print(wc(_in=ls("/etc", "-1"), "-l"))

提示
基础管道并不会异步传输数据流；内部的命令会阻塞到它结束运行，然后才会把数据传递给外部的命令。

默认情况下，接收管道输入的任何命令都要等待里面的命令运行结束。但是这一行为可以通过在内部命令中使用_piped这一特殊的动态参数改变，它可以告诉外部函数，不要等待内部函数完成才接收数据，而是异步、渐进地接收数据。建议先参考下面的例子。

高级

默认情况下，所有被管道传输的命令都会按顺序执行。这就意味着内部命令会优先执行，然后才把数据传给外部命令：

1
print(wc(_in=ls("/etc", "-1"), "-l"))

在以上例子中，ls执行，收集输出，然后传递给wc。这很适合简单的命令，但是对于你想要并行执行的命令，就不怎么样了。就拿下面的例子来说：

1
2
for line in tr(_in=tail("-f", "test.log"), "[:upper:]", "[:lower:]", _iter=True):
    print(line)

这段代码无法运行。因为tail -f命令永远也不会结束。你要做的是把tail的输出传递给tr，让它接收。这就凸显出_piped参数好用的地方了：

1
2
for line in tr(_in=tail("-f", "test.log", _piped=True), "[:upper:]", "[:lower:]", _iter=True):
    print(line)

这样才会正常运行，告诉tail -f它在管道中被调用了，应该把它的输出一行一行传给tr。默认情况下，_piped传递标准输出 ，但是你也可以轻松把它修改为传递错误输出，只需要让_piped="err"。

子命令详解

很多程序都有自己的命令子集。比如git (branch, checkout)，svn (update, status)，还有sudo（sudo后面的命令都是作为子命令运行的）。sh 可以通过取属性的方式来处理子命令：

1
2
3
4
5
6
7
8
9
from sh import git, sudo

# 解释为 "git branch -v"
print(git.branch("-v"))
print(git("branch", "-v")) # 相同的命令

# 解释为 "sudo /bin/ls /root"
print(sudo.ls("/root"))
print(sudo("/bin/ls", "/root")) # 相同的命令

子命令主要是一些让调用的程序看起来更符合逻辑的语法糖。

提示
如果你要用sudo执行子命令，一定要去看看使用Sudo。

默认参数

很多时候，你想覆写所有 sh 调用的命令的默认参数。比方说，假定你想让所有命令的输出都汇总到io_StringIO缓冲区。有点笨的方法是这样：

1
2
3
4
5
6
7
8
import sh
from io import StringIO

buf = StringIO()

sh.ls("/", _out=buf)
sh.whoami(_out=buf)
sh.ps("auxwf", _out=buf)

很明显，这很快就变成了重复书写的单调任务。幸运的是，我们可以创建执行上下文 (Execution context) 来允许我们设置所有从上下文中派生的命令的默认参数（sh 2.0 以后的版本删除了执行上下文，你应该使用参数烘焙）：

1
2
3
4
5
6
7
8
9
10
11
12
import sh
from io import StringIO

buf = StringIO()
# 这种方法已经被废弃了
#sh2 = sh(_out=buf)
# 使用参数烘焙
sh2 = sh.bake(_out=buf)

sh2.ls("/")
sh2.whoami()
sh2.ps("auxwf")

现在，任何从sh2运行的命令都会把它的输出传给StringIO实例buf。

执行上下文也可以被导入，就好像它是顶层的 sh 模块一样。

环境变量

_env这个特殊的动态参数允许你传递一个环境变量字典和它们的对应值：

1
2
import sh
sh.google_chrome(_env={"SOCKS_SERVER": "localhost:1234"})

_env会完全替换进程的环境变量，只有_env中的键值对会作为它的环境变量使用。如果你想以追加的形式添加新的环境变量的话，试试这样做：

1
2
3
4
5
6
7
import os
import sh

new_env = os.environ.copy()
new_env["SOCKS_SERVER"] = "localhost:1234"

sh.google_chrome(_env=new_env)

提示

  • 使用_cwd参数指定命令的工作目录。
  • 使用_uid参数指定命令的运行用户。

提示
要让一个环境变量对所有的 sh 命令适用，看看默认变量。

从标准输入中获取输入

使用函数的_in特殊动态参数可以让标准输入直接传递给进程：

1
print(cat(_in="test"))

任何从标准输入接收输入的命令都可以这么用：

1
print(tr("[:lower:]", "[:upper:]", _in="sh is awesome"))

不仅限于传递字符串，你还可以使用文件对象，queue.Queue，甚至任何可迭代类型（列表，集合，字典等等）：

1
2
stdin = ["sh", "is", "awesome"]
out = tr("[:lower:]", "[:upper:]", _in=stdin)

提示
如果你使用了队列，别忘了你可以用None标记 EOF。

‘With’上下文

命令可以在 Python 的with上下文中运行。最适合这么用的命令可能就是sudo或fakeroot了：

1
2
with sh.contrib.sudo:
    print(ls("/root"))

如果你需要在 with 上下文中运行命令并传递参数的话，比方说，给 sudo 设定一个 -p 提示语，你需要使用_with=True，这会让命令明白它在 with 上下文中，这样才能让它正常发挥作用：

1
2
with sh.contrib.sudo(k=True, _with=True):
    print(ls("/root"))

切换目录

举例来说，如果你希望临时切换工作目录，而不是使用os.chdir()命令切换脚本全局的工作目录的话，还可以使用：

1
2
with sh.pushd('/目录'):
    command()

使用Sudo

有两种使用sudo的方法：

终端输入

1
2
3
4
5
6
import sh

# 注意with
with sh.contrib.sudo:
    # 这条命令的权限是提升的
    command()

读取字符串

1
2
3
4
5
6
import sh

pass: str = input()
with sh.contrib.sudo(password=pass, _with=True):
    # 这条命令的权限是提升的
    command()

这种方法不太安全。当然，也可以给用户设置sudo的NOPASSWD权限。

其他技巧

缓冲区

默认情况下，标准输出和标准错误输出都是逐行缓冲的，但是可以通过两个变量修改这一行为：

  • _out_bufsize：为 0 时，标准输出为块缓冲，为 1 时，标准输出为行缓冲。
  • _err_bufsize：为 0 时，标准错误为块缓冲，为 1 时，标准错误为行缓冲。

调用脚本

sh.Command('脚本路径'[, '参数'])可以使你在Python脚本中调用其他的脚本。

测试通配符

sh.glob('路径通配符')可以测试通配符的匹配结果并返回文件名列表。它的功能实际上和glob.glob('路径通配符')相同。

运行 Shell 内置命令

sh.sh("-c", "命令")可以用于运行 Shell 的内置命令，而不是从 PATH 变量中寻找的命令。

类型检查

sh重载__getattr__(name: str)方法动态修改sys.modules的特性导致它会导致类型检查器触发reportAttributeAccessIssue问题，解决方法为创建一个Stub文件sh.pyi，内容为：

1
2
3
from typing import Any

def __getattr__(name: str) -> Any: ...