简化使用子进程和ssh的执行辅助工具。
项目描述
exec-helpers
简化使用子进程和ssh的执行辅助工具。为什么还需要另一个子进程包装器，为什么不是明确的paramiko？
从历史角度看，paramiko提供了良好的ssh客户端，但存在一些限制：您可以使用超时调用命令，但不能获取返回码，或者调用命令并等待返回码，但不能处理超时。
在大多数情况下，我们只需要一个简单的SSH客户端，它具有舒适的API调用，通过SSH代理进行调用，并检查返回代码/stderr。这个库提供了这种功能，具有无死锁的轮询和友好的结果对象（具有XML元素树、YAML、JSON、二进制或只是字符串的内置解码）。此外，这个库还提供了子进程调用的相同API，但有限制：不允许并行调用（以防止竞争条件）。
优点
命令执行期间对STDOUT和STDERR进行轮询 - 无死锁。
子进程和ssh使用相同的API。
免费软件：Apache许可证
自文档化代码：注释中的docstrings和类型
经过测试：请参阅顶部的徽章
支持多个Python版本
Python 3.8 Python 3.9 Python 3.10 Python 3.11 Python 3.12
此包包括
SSHClient - 历史上第一个助手，用于SSH连接。还提供了几个sFTP的API调用。
SSHAuth - 存储凭据的类。SSHClient不会直接存储凭据，而是使用SSHAuth。此类对象的实例可以在SSH连接对象之间复制，也用于execute_through_host。
Subprocess - 具有时限、轮询和几乎相同的API的subprocess.Popen包装器，与SSHClient相同（除了特定标志，如子进程的cwd和ssh的get_tty）。
async_api.Subprocess - 与Subprocess助手相同，但与asyncio一起工作。..注意：对于Windows，应使用ProactorEventLoop或另一个非标准事件循环！
ExecResult - 存储执行结果的类。包含退出代码、stdout、stderr，并提供解码为JSON、YAML、XML（和LXML）元素树、字符串、bytearray和简短字符串（最多7行）的getter。
ExitCodes - 标准Linux退出代码的枚举器。也提供了由信号代码产生的BASH返回代码。
安装
标准：pip install exec-helpers 额外
yaml - 安装PyYaml以解码YAML（PyYAML是主要解码器，ruamel.YAML也作为后备支持。）
xml - 安装defusedxml以安全地将XML解析为xml.etree.ElementTree.Element。
lxml - 安装lxml以进行高级XML解析。可能不安全。
ALL_FORMATS (all-formats) - 安装所有解析器。当添加新的解析器时，也将支持。
使用
SSHClient
可以不构建特定对象来初始化SSHClient的基本初始化
client = exec_helpers.SSHClient(host, username="username", password="password")如果ssh代理正在运行，则密钥将由paramiko自动收集，但如果密钥位于特定位置，则应手动加载并作为paramiko.PKey的可迭代对象提供。
对于高级情况或凭据的重用，应使用SSHAuth对象。可以从连接对象通过属性auth收集。
从头开始创建
auth = exec_helpers.SSHAuth(
username='username', # str | None
password='password', # str | None
key=None, # type: paramiko.PKey | None
keys=None, # type: Iterable[paramiko.PKey] | None
key_filename=None, # type: list[str] | None
passphrase=None, # str | None
)密钥是主要连接密钥（始终首先尝试），密钥是备用密钥。密钥文件名是密钥的文件名或文件名列表，应加载这些密钥。密码是密钥的备用密码，如果它与主要密码不同。如果主要密钥现在正确用于用户名，则尝试备用密钥，如果在找到正确密钥后，则它成为主要密钥。如果没有找到有效的密钥，则使用密码，并将主密钥设置为None。
提供了上下文管理器，连接在退出上下文时关闭并释放锁。
子进程
提供了上下文管理器，子进程在退出上下文时被终止并释放锁。
基本方法
主要方法包括 execute、check_call 和 check_stderr，用于简单的执行、执行并检查返回码以及执行、检查返回码和检查标准错误输出是否为空。这些方法在 SSHClient 和 Subprocess 中几乎相同，除了特定的标志。
result: ExecResult = helper.execute(
command, # type: str | Iterable[str]
verbose=False, # type: bool
timeout=1 * 60 * 60, # type: int | float | None
# Keyword only:
log_mask_re=None, # str | None
stdin=None, # type: bytes | str | bytearray | None
open_stdout=True, # type: bool
log_stdout=True, # type: bool
open_stderr=True, # type: bool
log_stderr=True, # type: bool
**kwargs
)result: ExecResult = helper.check_call(
command, # type: str | Iterable[str]
verbose=False, # type: bool
timeout=1 * 60 * 60, # type: type: int | float | None
error_info=None, # str | None
expected=(0,), # type: Iterable[int | ExitCodes]
raise_on_err=True, # type: bool
# Keyword only:
log_mask_re=None, # str | None
stdin=None, # type: bytes | str | bytearray | None
open_stdout=True, # type: bool
log_stdout=True, # type: bool
open_stderr=True, # type: bool
log_stderr=True, # type: bool
exception_class=CalledProcessError, # type[CalledProcessError]
**kwargs
)result: ExecResult = helper.check_stderr(
command, # type: str | Iterable[str]
verbose=False, # type: bool
timeout=1 * 60 * 60, # type: type: int | float | None
error_info=None, # str | None
raise_on_err=True, # type: bool
# Keyword only:
expected=(0,), # Iterable[int | ExitCodes]
log_mask_re=None, # str | None
stdin=None, # type: bytes | str | bytearray | None
open_stdout=True, # type: bool
log_stdout=True, # type: bool
open_stderr=True, # type: bool
log_stderr=True, # type: bool
exception_class=CalledProcessError, # type[CalledProcessError]
)result: ExecResult = helper( # Lazy way: instances are callable and uses `execute`.
command, # type: str | Iterable[str]
verbose=False, # type: bool
timeout=1 * 60 * 60, # type: int | float | None
# Keyword only:
log_mask_re=None, # str | None
stdin=None, # type: bytes | str | bytearray | None
open_stdout=True, # type: bool
log_stdout=True, # type: bool
open_stderr=True, # type: bool
log_stderr=True, # type: bool
**kwargs
)如果不需要标准输出或标准错误，可以通过使用带有标志 open_stdout=False 和 open_stderr=False 的 **kwargs 来禁用这些FIFO管道。
下一级命令使用较低级别，kwargs会转发，因此期望的退出码从 check_stderr 转发。特定的实现标志始终通过kwargs设置。
如果需要从日志中隐藏命令的一部分，可以通过实例或命令设置 log_mask_re 属性。所有正则表达式匹配的组都将替换为 ‘<*masked*>’。
result: ExecResult = helper.execute(
command="AUTH='top_secret_key'; run command", # type: str | Iterable[str]
verbose=False, # type: bool
timeout=1 * 60 * 60, # type: Optional[int]
log_mask_re=r"AUTH\s*=\s*'(\w+)'" # str | None
)result.cmd 将等于 AUTH=’<*masked*>’; run command
ExecResult
执行结果对象有一组有用的属性
cmd - 命令
exit_code - 命令返回码。如果可能，使用Linux的枚举器进行解码 -> 使用它。
ok -> bool。命令返回码为0（EX_OK）。
stdin -> str。stdin的文本表示。
stdout -> tuple[bytes]。原始stdout输出。
stderr -> tuple[bytes]。原始stderr输出。
stdout_bin -> bytearray。二进制stdout输出。
stderr_bin -> bytearray。二进制stderr输出。
stdout_str -> str。输出的文本表示。
stderr_str -> str。输出的文本表示。
stdout_brief -> str。最多7行stdout（如果超过7行，则为前3行和后3行）。
stderr_brief -> str。最多7行stderr（如果超过7行，则为前3行和后3行）。
stdout_json - 将STDOUT解码为JSON。
stdout_yaml - 将STDOUT解码为YAML。只有当安装了 PyYAML 或 ruamel.YAML 库时才可用。（附加：yaml）
stdout_xml - 使用 defusedxml 库将STDOUT解码为XML到 ElementTree。只有当安装了 defusedxml 库时才可用。（附加：xml）
stdout_lxml - 使用 lxml 库将STDOUT解码为XML到 ElementTree。只有当安装了 lxml 库时才可用。（附加：lxml）。可能不安全。
timestamp -> Optional(datetime.datetime)。接收退出码的时间戳。
SSHClient 特定
SSHClient 命令支持 get_pty 标志，该标志在远程端启用PTY打开。可以通过关键字参数设置PTY宽度和高度，尺寸以像素为单位始终为0x0。
如果不会产生大量输出，可以在多个主机上并行调用命令。
results: dict[tuple[str, int], ExecResult] = SSHClient.execute_together(
remotes, # type: Iterable[SSHClient]
command, # type: str | Iterable[str]
timeout=1 * 60 * 60, # type: type: int | float | None
expected=(0,), # type: Iterable[int | ExitCodes]
raise_on_err=True, # type: bool
# Keyword only:
stdin=None, # type: bytes | str | bytearray | None
open_stdout=True, # type: bool
open_stderr=True, # type: bool
log_mask_re=None, # str | None
exception_class=ParallelCallProcessError # type[ParallelCallProcessError]
)
results # type: dict[tuple[str, int], exec_result.ExecResult]结果是一个字典，键 = (主机名，端口)，值 = 结果。默认情况下，如果任何远程返回意外退出码，则 execute_together 将引发异常。
要使用当前作为代理打开新连接，可以使用可访问的方法 proxy_to。基本用法示例
conn: SSHClient = client.proxy_to(host, username="username", password="password")可以使用 execute_through_host 方法通过SSH主机执行
result: ExecResult = client.execute_through_host(
hostname, # type: str
command, # type: str | Iterable[str]
# Keyword only:
auth=None, # type: SSHAuth | None
port=22, # type: int
timeout=1 * 60 * 60, # type: type: int | float | None
verbose=False, # type: bool
stdin=None, # type: bytes | str | bytearray | None
open_stdout=True, # type: bool
log_stdout=True, # type: bool
open_stderr=True, # type: bool
log_stderr=True, # type: bool
log_mask_re=None, # str | None
get_pty=False, # type: bool
width=80, # type: int
height=24 # type: int
)其中主机名是目标主机名，auth是目标主机的备用凭据。
SSH客户端通过上下文管理器实现快速的sudo支持
命令将以sudo强制执行，与客户端设置独立，用于正常使用。
with client.sudo(enforce=True):
...命令将以无sudo方式独立于客户端设置运行，用于正常使用。
with client.sudo(enforce=False):
...“永久性客户端设置”
client.sudo_mode = mode # where mode is True or FalseSSH客户端支持sFTP以处理远程文件。
with client.open(path, mode='r') as f:
...为快速检查远程路径，可用方法如下
exists(path) -> bool
>>> conn.exists('/etc/passwd')
Truestat(path) -> paramiko.sftp_attr.SFTPAttributes
>>> conn.stat('/etc/passwd')
<SFTPAttributes: [ size=1882 uid=0 gid=0 mode=0o100644 atime=1521618061 mtime=1449733241 ]>
>>> str(conn.stat('/etc/passwd'))
'-rw-r--r-- 1 0 0 1882 10 Dec 2015 ?'isfile(path) -> bool
>>> conn.isfile('/etc/passwd')
Trueisdir(path) -> bool
>>> conn.isdir('/etc/passwd')
False附加（非标准）助手
mkdir(path: str) - 执行mkdir -p path
rm_rf(path: str) - 执行rm -rf path
upload(source: str, target: str) - 使用sFTP从源上传文件到目标。
download(destination: str, target: str) - 使用sFTP从目标下载文件到目的地。
子进程特定
关键字参数
cwd - 工作目录。
env - 环境变量字典。
async_api.子进程特定
所有标准方法都是协程。还可用异步上下文管理器。
示例
async with helper:
result: ExecResult = await helper.execute(
command, # type: str | Iterable[str]
verbose=False, # type: bool
timeout=1 * 60 * 60, # type: int | float | None
**kwargs
)测试
该包 exec-helpers 的主要测试机制是使用 tox。可以通过 tox -l 收集可用环境。
CI系统
为代码检查，并行使用几个CI系统。
GitHub actions: 用于检查：PEP8、pylint、bandit、安装可能性以及单元测试。
下载文件
下载适用于您平台的文件。如果您不确定选择哪个，请了解有关安装包的更多信息。
