Python sh库完整翻译

发表于 更新于

什么是 sh

sh 是一个成熟的 subprocess 替代品，适用于 Python2.6 - 3.8, 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 。

安装

pip install sh

快速参考

传参

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")

阅读更多

管道

sh.wc(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
# 解释为 "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"
adduser("amoffat", system=True, shell="/bin/bash", no_create_home=True)

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

退出代码与异常

众所周知，正常进程结束后会以 0 代码退出。这个代码会通过变量RunningCommand.exit_code返回：

1
2
output = ls("/")
print(output.exit_code) # 应该是0

如果一个进程中止了，那返回代码就不是0，一个异常也会自动生成。你可以捕获这些特定的返回码，也可以干脆通过错误基类ErrorReturnCode把它们全部捕获：

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

提示
你可以通过_ok_code变量自定义代表异常的返回码

信号

信号会在进程因接收到信号而终止时抛出。此时抛出的异常类型为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

重定向详解

sh 可以重定向进程的标准输出流和错误输出流到各种不同类型的目标，只需要使用_out和_err这两个特殊的动态参数。

重定向到文件名

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

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

提示
那我咋追加到文件呢？

类文件对象

你也可以重定向到任何支持.write(data)方法的对象，比如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信号，这意味这当它们的控制进程 (如果有控制终端的话，就是 Session leader ) 退出的话，他们就不会收到系统的信号了。但是由于 sh 命令默认情况下会运行它们自己的进程，这也就是说他们自己就是 Session leader，无视SIGHUP一般情况下并没有影响。唯一会产生影响的情况是你使用了_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]

参数烘焙 (Bake)

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
15
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))

管道详解

基础

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

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

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

提示
基础管道并不会异步传输数据流；内部的命令会阻塞到它结束运行，然后才会把数据传递给外部的命令。
默认情况下，接收管道输入的任何命令都要等待里面的命令运行结束。但是这一行为可以通过在内部命令中使用_piped这一特殊的动态参数改变，它可以告诉外部函数，不要等待内部函数完成才接收数据，而是递增的接收数据。建议先参考下面的例子。

高级

默认情况下，所有被管道传输的命令都会按顺序执行。这就意味着内部命令会优先执行，然后才把数据传给外部命令：
print(wc(ls("/etc", "-1"), "-l"))
在以上例子中，ls执行，收集输出，然后传递给wc。这很适合简单的命令，但是对于你想要并行执行的命令，就不怎么样了。就拿下面的例子来说：

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

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

1
2
for line in tr(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")) # 相同的命令

# resolves to "sudo /bin/ls /root"
print(sudo.ls("/root"))
print(sudo("/bin/ls", "/root")) # 相同的命令

子命令主要是一些让调用的程序看起来更符合逻辑的语法糖。
提示
如果你要用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) 来允许我们设置所有从上下文中派生的命令的默认参数：

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

buf = StringIO()
sh2 = sh(_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)

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

从标准输入中获取输入

使用命令的_in特殊动态参数可以让标准输入直接传递给进程：
print(cat(_in="test"))
任何从标准输入接收输入的命令都可以这么用：
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"))