Skip to main content

简化子流程

项目描述

特拉维斯 CI 代码覆盖率

Subby 是一个小型 Python 库,旨在简化子流程的使用。Subby 类似于 delegator.py 但它添加了一些附加功能并排除了其他功能(例如不pexpect支持)。

要求

唯一的要求是 python 3.6+。没有其他第 3 方运行时依赖项。pytestcoverage包是测试所必需的。

安装

pip install subby

用法

Subby 的主要接口是run函数。它需要一个命令列表并执行它们。如果有多个命令,它们被链接(即管道)在一起。

import subby

# We can pass input to the stdin of the command as bytes
input_str = "foo\nbar"

# The following three commands are equivalent; each returns a
# `Processes` object that can be used to inspect and control
# the process(es).
p1 = subby.run([["grep foo", "wc -l"]], stdin=input_str)
p2 = subby.run(("grep foo", "wc -l"), stdin=input_str)
p3 = subby.run("grep foo | wc -l", stdin=input_str)

# The `done` property tells us whether the processes have finished
assert p1.done and p2.done and p3.done

# The `output` property provides the output of the command
assert p1.output == p2.output == p3.output == "1"

原始模式

默认情况下,文本 I/O 用于 stdin/stdout/stderr。您可以通过传递来使用原始 I/O(字节)mode=bytes

import subby

assert b"1" == subby.run(
    "grep foo | wc -l", stdin="foo\nbar", mode=bytes
).output

非阻塞进程

默认情况下,该run功能会阻塞,直到进程完成运行。可以通过传递来更改此行为block=False,在这种情况下,调用者负责检查状态和/或Processes.block()手动调用方法。

import subby
import time

p = subby.run("sleep 10", block=False)
for i in range(5):
    if p.done:
        break
    else:
        time.sleep(1)
else:
    # A timeout can be used to kill the process if it doesn't
    # complete in a certain amount of time. By default, block()
    # raises an error if the return code is non-zero.
    p.block(timeout=10, raise_on_error=False)
    
    # The process can also be killed manually.
    p.kill()

# The `Processes.ok` property is True if the processes have
# finished and the return code is 0.
if not p.ok:
    # The `Processes.output` and `Processes.error` properties
    # provide access to the process stdout and stderr.
    print(f"The command failed: stderr={p.error}")

方便的方法

还有一个方便的方法 ,sub相当于runmode=strand调用block=True并返回结果对象的output属性(stdout) 。Processes

import subby

assert subby.sub("grep foo | wc -l", stdin="foo\nbar") == "1"

标准输入/标准输出/标准错误

Subby 支持标准输入、标准输出和标准错误的几种不同类型的参数:

  • 文件:指定为pathlib.Path; 对于标准输入,从文件中读取内容,而对于标准输出/标准错误,内容被写入文件(因此无法通过output/error属性获得)。
  • 字节字符串:对于标准输入,字节被写入一个临时文件,该文件被传递给进程标准输入。
  • StdType枚举 提供的值之一:
    • PIPE:用于 stdout/stderr,subprocess.PIPE使调用者可以直接访问进程 stdout/stderr 流。
    • BUFFER:对于stdout/stderr,使用一个临时文件,并且在进程完成后通过output/属性使内容可用。error
    • SYS:stdin/stdout/stderr 是从主进程(即sys.stdin/sys.stdout/sys.stderr流)传递过来的。

默认情况下,会捕获链中所有进程的 stderr 流(您可以通过传递capture_stderr=False来禁用它run())。

import subby
p = subby.run("echo -n hi | tee /dev/stderr | tee /dev/stderr")
assert p.output == b"hi"
assert p.get_all_stderr() == [b"", b"hi", b"hi"]

日志记录

默认情况下,所有执行的命令都会被记录(使用 loglevel INFO)。echo=False您可以通过传递给来禁用此行为run()

import subby
subby.run("touch foo")  # Echoes "touch foo" to the log with level INFO
subby.run("login -p mypassword", echo=False)  # Does not echo mypassword

返回码

默认情况下,Subby 将返回代码0视为成功,将所有其他返回代码视为失败。在某些情况下,这不是所需的行为。一个著名的例子是grep,它有一个1没有匹配行的返回码。要忽略其他返回码,请将allowed_return_codes关键字参数设置为run()

import subby
subby.run("echo foo | grep bar")  # Raises CalledProcessError
subby.run("echo foo | grep bar", allowed_return_codes=(0, 1))

贡献

Subby 被认为在很大程度上功能完整,但如果您发现错误或有改进建议,请提交问题(或者更好的是,拉取请求)。

致谢

Subby 的灵感来自delegator.py

Subby 最初是作为dxpy.sugar包的一部分编写的,但因为它(希望)更普遍有用,所以它作为一个单独的包提供。@Damien-Black@msimbirsky贡献了代码和评论。

项目详情


下载文件

下载适用于您平台的文件。如果您不确定要选择哪个,请了解有关安装包的更多信息。

源分布

subby-0.1.7.ta​​r.gz (16.6 kB 查看哈希)

已上传 source

内置分布

subby-0.1.7-py3-none-any.whl (15.4 kB 查看哈希)

已上传 py3