子流程简化

subb的Python项目详细描述


Travis CICode Coverage

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(字节)。在

^{pr2}$

非阻塞过程

默认情况下,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流)传递的。在

默认情况下,捕获链中所有进程的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}贡献了代码和评论。在

欢迎加入QQ群-->: 979659372 Python中文网_新手群

推荐PyPI第三方库


热门话题
java如何删除列表中的重复项并将列表复制到另一个列表中   Java如何在非UI应用程序中检测tab键按下?   java无法在未调用Looper的线程内创建处理程序。警报对话框线程上的prepare()   java Jaxb将XML转换为ElementNSImpl类型,而不是正确的类型   基于java标记的过滤实现Firestore   我们可以使用Java和Eclipse使用YouTube分析Api吗?   JavaSpring注释在多个选项卡中同时打开多个视图   java Hibernate不使用JBOSS检索数据   矩阵如何从Apache Commons数学Java库中找到伪逆   对象Java打印“null”作为字符串引用!=使无效   java中的图形慢映射   java用户界面:操作栏和上下文操作栏,新趋势   java Infinispan,版本化操作返回不正确的结果   java调整保存在父JFrame中的JInternalFrame的大小,并且该父JFrame使用ComponentMover API注册   java如何防止会话属性在服务器上持久化?   java如何为SpringBoot 2创建自定义嵌入式servlet容器   使用Robolectric进行单元测试时,在片段类内调用getString()时发生java IllegalStateException