相关文章推荐
傻傻的馒头  ·  Nginx下访问.html ...·  1 月前    · 

本文主要参考和对官方文档进行二次浓缩而成

在使用 Python 编写程序的时候,我们常常会使用 Linux 系统下面的命令,比如需要执行一个 ifconfig 命令等。这时候,我们常常需要使用 os.popen os.system commands subprocess 来完成,但是使用起来还是很不方便,有违和感。但是通过 sh 可以像调用函数一样调用 Linux 下系统命令。

1. sh 库安装方式

sh 库将系统的命令动态映射到 Python 函数

  • [1] 安装方法
  • # pip
    $ pip install sh
    # pyenv
    $ pyenv install sh
    # poetry
    $ poetry install sh
    
  • [2] 主要操作
  • In [9]: output.process Out[9]: <Process 60644 [b'/usr/bin/command', b'who']> In [10]: output.next() Out[10]: 'escape console Sep 29 09:10'

    2. sh 库实现原理

    具体的实现可以参考源码

    在 sh 库中,其并没有重新实现了类似 lssudo 这样的 Linux 命令,而是使用了一种我们在 Python 编写中不常用的 ModuleType 类型来实现的。

    >>> import types
    >>> types.ModuleType
    <type 'module'>
    

    在编程中,我们所导入的每个模块都是这种类型。这里,我们将导入的模块替换为我们自己定义的模块,即可。当我们去访问模块不存在的属性时,会自动调用定义的 __getattr__ 魔术方法,是不是很巧妙?从这我们知道怎么从一个 moduleimport 出不存在的函数或者类了,那么它是怎么把 import 的函数绑定到系统中呢?这个可以看下 源码,然后思考思考。

    import sys
    from types import ModuleType
    class Tenv(ModuleType):
        def __init__(self, self_module):
            self.self_module = self_module
        def __getattr__(self, name):
            return f'{name} to -> test'
    if __name__ == "__main__":
    else:
        self = sys.modules[__name__]
        sys.modules[__name__] = Tenv(self)
    
    >>> import test
    >>> print test.ls
    ('ls' to -> test)
    

    3. sh 库使用技巧

    这里主要介绍简单的快速上手的相关操作

  • [1] 默认参数
  • 很多时候,我们希望覆盖通过 sh 启动的所有命令的默认参数。

    import sh
    from io import StringIO
    buf = StringIO()
    sh2 = sh(_out=buf)
    from sh2 import ls, whoami, ps
    ls("/")
    whoami()
    ps("auxwf")
    
  • [2] 传递位置参数
  • 当向命令传递多个参数时,每个参数必须是单独的字符串。

    from sh import tar
    tar("cvf", "/tmp/test.tar", "/my/home/directory/")
    
  • [3] 传递关键字参数
  • sh 支持将短格式 -a 和长格式 --arg 参数用作关键字参数。

    from sh import curl
    curl("http://duckduckgo.com/", o="page.html", silent=True)
    curl("http://duckduckgo.com/", "-o", "page.html", "--silent")
    
  • [4] 退出码
  • 正常进程在执行完成之后,会以退出代码 0 退出,可以通过 RunningCommand.exit_code 看到。

    from sh import ls
    output = ls("/")
    print(output.exit_code) # should be 0
    
    from sh import ls
    try:
        print(ls("/some/non-existant/folder"))
    except ErrorReturnCode_2:
        print("folder doesn't exist!")
        create_the_folder()
    except ErrorReturnCode:
        print("unknown error")
    
  • [5] 信号
  • 当进程因信号而终止时,就会发出信号,在这种情况下会引发 SignalException 异常。

    import sh
    try:
        p = sh.sleep(3, _bg=True)
        p.kill()
    except sh.SignalException_SIGKILL:
        print("killed")
    
  • [6] 重定向
  • sh 库中,我们可以使用 _out_err 来重定向默认的 STDOUTSTDERR

    import sh
    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)
    
  • [7] 自定义模式
  • sh 可以将参数固定之后,转化为新的命令,其本质上就像使用 functools.partial() 一样。

    my_ls = sh.ls.bake("-l")
    my_ls("/tmp")
    # equivalent
    sh.ls("-l", "/tmp")
    
  • [8] 管道
  • Bash 的管道是使用函数组合来执行的,只要将一个命令作为输入传递给另一个命令。

    print(wc(ls("/etc", "-1"), "-l"))
    print(sort(du(glob("*"), "-sb"), "-rn"))
    for line in tr(tail("-f", "test.log"), "[:upper:]", "[:lower:]", _iter=True):
        print(line)
    
  • [9] 子命令
  • 许多工具都有自己的子命令,例如 gitsudo,同样 sh 也是支持的。

    # equivalent
    sh.git("show", "HEAD")
    sh.git.show("HEAD")
    
  • [10] 后台处理
  • 默认情况下,每个正在运行的命令在完成之前都会阻塞。如果有一个长时间运行的命令,可以使用 _bg=True 将其放在后台。

    # blocks
    sleep(3)
    print("...3 seconds later")
    # doesn't block
    p = sleep(3, _bg=True)
    print("prints immediately!")
    p.wait()
    print("...and 3 seconds later")
    
  • [11] 输出回调
  • sh 库中支持使用 _out_err 来处理正常和错误输出的回调函数。

    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()
    
    from sh import tail
    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)
    p.wait()
    
    from sh import ssh
    def ssh_interact(line, stdin):
        line = line.strip()
        print(line)
        if line.endswith("password:"):
            stdin.put("correcthorsebatterystaple")
    ssh("10.10.10.100", _out=ssh_interact)
    
    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]
    
  • [12] 环境变量
  • 使用 _env 变量允许我们传递环境变量及其对应值的字典。

    import sh
    sh.google_chrome(_env={"SOCKS_SERVER": "localhost:1234"})
    
    import os
    import sh
    new_env = os.environ.copy()
    new_env["SOCKS_SERVER"] = "localhost:1234"
    sh.google_chrome(_env=new_env)
    
  • [13] 输入输出
  • 通过 _in 变量给 STDIN 进程直接发送信息。

    print(cat(_in="test"))
    print(tr("[:lower:]", "[:upper:]", _in="sh is awesome"))
    stdin = ["sh", "is", "awesome"]
    out = tr("[:lower:]", "[:upper:]", _in=stdin)
    
  • [14] 上下文
  • 命令可以在带有上下文的 Python 中运行,常用的可能是 sudo 命令。

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

    4. sh 库参考内容

    主要介绍一些特殊的参数、错误异常类型等

  • [1] 特殊关键字 - 控制输出
  •