简化子流程
项目描述
Subby 是一个小型 Python 库,旨在简化子流程的使用。Subby 类似于 delegator.py ,但它添加了一些附加功能并排除了其他功能(例如不pexpect
支持)。
要求
唯一的要求是 python 3.6+。没有其他第 3 方运行时依赖项。pytest
和coverage
包是测试所必需的。
安装
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
相当于run
用mode=str
and调用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
流)传递过来的。
- PIPE:用于 stdout/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贡献了代码和评论。
项目详情
下载文件
下载适用于您平台的文件。如果您不确定要选择哪个,请了解有关安装包的更多信息。