程序员友好的子进程包装器
项目描述
executor包是Python的subprocess模块的一个简单包装器,它使得在UNIX系统上处理子进程变得非常容易,它能够正确地转义参数并进行错误检查。
使用面向对象的接口执行命令,使用合理的默认值(可自定义且文档齐全)。
还支持并发执行一组命令,称为“命令池”。并发级别可以自定义,当然也支持本地和远程命令。
该包目前在Python 2.7、3.5、3.6、3.7、3.8和PyPy上进行测试。有关使用说明,请参阅以下章节和文档。
安装
《executor》包可在PyPI上获取,这意味着安装应该非常简单。
$ pip install executor
安装Python包的方法有很多(例如,按用户站点的包目录,虚拟环境或全局安装),我无意在此讨论这些方法,所以如果你感到害怕,请阅读你的选项后再回来阅读这些说明 ;-).
使用
使用《executor》包有两种方式:作为命令行程序executor和作为Python API。下面将描述命令行界面,同时也有一些Python API的简单用例。
命令行
用法: executor [OPTIONS] COMMAND …
基于同名的Python包的命令行子进程管理工具。该“executor”程序支持超时、动态启动延迟(调整因子)和独占锁定。
可以将“executor”视为“flock”和“timelimit”程序的组合,并添加一些额外的特性(即动态启动延迟和UNIX平台上的集成系统日志)。
支持选项
选项 |
描述 |
---|---|
-t, --timeout=LIMIT |
设置给定命令将被中止的时间。默认情况下,LIMIT按秒计算。您也可以使用后缀“s”(秒)、“m”(分钟)、“h”(小时)或“d”(天)。 |
-f, --fudge-factor=LIMIT |
此选项控制动态启动延迟(调整因子),当您希望周期性任务在给定间隔内运行一次,但确切时间不重要时很有用。有关LIMIT的可接受值的更多信息,请参阅--timeout选项,此数字指定在运行命令之前睡眠的最长时间(最小值为零,否则您可以在命令行中包含“sleep N && …”命令 :-)。 |
-e, --exclusive |
使用进程间锁文件来保证executor永远不会并发运行外部命令。有关定制的阻塞/非阻塞行为的更多信息,请参阅--lock-timeout选项。要自定义锁文件的名称,可以使用--lock-file选项。 |
-T, --lock-timeout=LIMIT |
默认情况下,executor会尝试获取锁,如果失败,则使用非零退出代码退出。此选项可用于启用阻塞行为。有关LIMIT的可接受值的更多信息,请参阅--timeout选项。 |
-l, --lock-file=NAME |
自定义锁文件的名称。默认情况下,这是外部命令的基本名称,因此如果您正在运行像“bash”或“python”这样的通用命令,您可能希望更改此设置 :-). |
-v, --verbose |
增加日志详细程度(可重复)。 |
-q, --quiet |
减少日志详细程度(可重复)。 |
-h,--help |
显示此信息并退出。 |
Python API
以下是一些关于 execute() 函数如何灵活的示例。有关(大量)其他用例,请参考 Read the Docs 上的 API 文档。
检查状态码
默认情况下,外部命令的状态码作为布尔值返回
>>> from executor import execute >>> execute('true') True
如果外部命令以非零状态码退出,将引发异常,这使得正确行事变得容易(无需编写大量重复代码,也无需忘记检查外部命令的状态码)
>>> execute('false') Traceback (most recent call last): File "executor/__init__.py", line 124, in execute cmd.start() File "executor/__init__.py", line 516, in start self.wait() File "executor/__init__.py", line 541, in wait self.check_errors() File "executor/__init__.py", line 568, in check_errors raise ExternalCommandFailed(self) executor.ExternalCommandFailed: External command failed with exit code 1! (command: bash -c false)
ExternalCommandFailed 异常公开了 command 和 returncode 属性。如果您知道某个命令很可能以非零状态码退出,并且希望 execute() 仅返回布尔值,您可以这样做
>>> execute('false', check=False) False
提供输入
以下是向外部命令提供输入的方法
>>> execute('tr a-z A-Z', input='Hello world from Python!\n') HELLO WORLD FROM PYTHON! True
获取输出
获取外部命令的输出同样非常简单
>>> execute('hostname', capture=True) 'peter-macbook'
以 root 用户运行命令
以超级用户权限执行命令也非常简单
>>> execute('echo test > /etc/hostname', sudo=True) [sudo] password for peter: ********** True >>> execute('hostname', capture=True) 'test'
启用日志记录
如果您想知道如何使用 sudo 前缀上述命令会有何帮助,请看以下操作方式
>>> import logging >>> logging.basicConfig() >>> logging.getLogger().setLevel(logging.DEBUG) >>> execute('echo peter-macbook > /etc/hostname', sudo=True) DEBUG:executor:Executing external command: sudo bash -c 'echo peter-macbook > /etc/hostname'
运行远程命令
要使用 SSH 在远程系统上运行命令,您可以使用 RemoteCommand 类,其工作原理如下
>>> from executor.ssh.client import RemoteCommand >>> cmd = RemoteCommand('localhost', 'echo $SSH_CONNECTION', capture=True) >>> cmd.start() >>> cmd.output '127.0.0.1 57255 127.0.0.1 22'
并发运行远程命令
foreach() 函数封装了 RemoteCommand 和 CommandPool 类,使得在多个主机上并发运行远程命令变得非常简单
>>> from executor.ssh.client import foreach >>> from pprint import pprint >>> hosts = ['127.0.0.1', '127.0.0.2', '127.0.0.3', '127.0.0.4'] >>> commands = foreach(hosts, 'echo $SSH_CONNECTION') >>> pprint([cmd.output for cmd in commands]) ['127.0.0.1 57278 127.0.0.1 22', '127.0.0.1 52385 127.0.0.2 22', '127.0.0.1 49228 127.0.0.3 22', '127.0.0.1 40628 127.0.0.4 22']
联系
executor 的最新版本可在 PyPI 和 GitHub 上找到。文档托管在 Read the Docs 上,包括一个 变更日志。有关错误报告,请在 GitHub 上创建问题。如果您有任何问题,建议等,请随时给我发电子邮件至 peter@peterodding.com。
许可证
此软件受 MIT 许可证 许可。
© 2020 Peter Odding。
项目详情
下载文件
下载适用于您平台的文件。如果您不确定选择哪个,请了解更多关于安装包的信息。