子流程简化
subb的Python项目详细描述
Subby是一个小型Python库,其目标是简化子进程的使用。Subby类似于delegator.py,但它添加了一些附加特性,并排除了其他特性(例如不支持pexpect
)。在
要求
唯一的要求是python3.6+。没有其他第三方运行时依赖项。测试需要pytest
和{
安装
pip install subby
使用
Subby的主要接口是run
函数。它获取命令列表并执行它们。如果有多个命令,它们将被链接(即管道)在一起。在
importsubby# We can pass input to the stdin of the command as bytesinput_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 finishedassertp1.doneandp2.doneandp3.done# The `output` property provides the output of the commandassertp1.output==p2.output==p3.output=="1"
原始模式
默认情况下,文本I/O用于stdin/stdout/stderr。您可以通过传递mode=bytes
来使用原始I/O(字节)。在
非阻塞过程
默认情况下,run
函数将一直阻塞,直到发现进程正在运行。这个行为可以通过传递block=False
来改变,在这种情况下,调用者负责检查状态和/或手动调用Processes.block()
方法。在
importsubbyimporttimep=subby.run("sleep 10",block=False)foriinrange(5):ifp.done:breakelse: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.ifnotp.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
,相当于用mode=str
和{Processes
对象的output
属性(stdout)。在
importsubbyassertsubby.sub("grep foo | wc -l",stdin="foo\nbar")=="1"
标准/标准/标准偏差
Subby支持stdin、stdout和stderr的几种不同类型的参数:
- 文件:指定为
pathlib.Path
;对于stdin,内容从文件中读取,而对于stdout/stderr,内容被写入文件(因此不能通过output
/error
属性使用)。在 - 字节字符串:对于stdin,字节被写入一个临时文件,该文件被传递给进程stdin。在
StdType
枚举提供的值之一:- 管道:对于stdout/stderr,使用
subprocess.PIPE
,这使调用者可以直接访问进程stdout/stderr流。在 - 缓冲区:对于stdout/stderr,使用临时文件,在进程完成后,内容通过^{cd14>}/^{cd18>}属性提供。在
- SYS:stdin/stdout/stderr是从主进程(即
sys.stdin/sys.stdout/sys.stderr
流)传递的。在
- 管道:对于stdout/stderr,使用
默认情况下,捕获链中所有进程的stderr流(您可以通过将capture_stderr=False
传递给run()
来禁用此功能)。在
importsubbyp=subby.run("echo -n hi | tee /dev/stderr | tee /dev/stderr")assertp.output==b"hi"assertp.get_all_stderr()==[b"",b"hi",b"hi"]
日志
默认情况下,所有执行的命令都会被记录(带有loglevel信息)。您可以通过将echo=False
传递给run()
来禁用此行为。在
importsubbysubby.run("touch foo")# Echoes "touch foo" to the log with level INFOsubby.run("login -p mypassword",echo=False)# Does not echo mypassword
返回码
{28>在默认情况下,将所有的成功^视为返回的子代码。在某些情况下,这不是期望的行为。一个著名的例子是grep
,当没有行匹配时,它的返回码是1
。要忽略其他返回码,请将allowed_return_codes
关键字参数设置为run()
。在
importsubbysubby.run("echo foo | grep bar")# Raises CalledProcessErrorsubby.run("echo foo | grep bar",allowed_return_codes=(0,1))
贡献
Subby被认为基本上已经完成了功能,但是如果你发现了一个bug或者有改进的建议,请提交一个问题(或者更好的是,请求拉取)。在
致谢
Subby的灵感来自于delegator.py。在
Subby最初是作为dxpy.sugar包的一部分编写的,但是由于它(希望)更普遍地有用,它被作为一个单独的包提供。@Damien-Black和{a7}贡献了代码和评论。在
- 项目
标签: