跳转至

项目分析与简单实现

项目实现

首先,我们要明确的是任务软件都是先实现一个简单的版本,然后再逐步完善功能,同样我们这里也是这样,先实现一个核心的功能,然后再逐步完善,核心功能就是通过 ssh 连接到远程服务器执行命令,所以接下来我们将实现这个功能.

首先创建一个新的项目目录 project-ansible ,然后再目录下创建一个名为 demoenv 的虚拟环境

$ mkdir project-ansible
$ cd project-ansible
$ python3 -m venv demoenv # 创建虚拟环境
$ source demoenv/bin/activate # 激活虚拟环境

注意 vscode 中,选择该虚拟环境的 Python 解释器

配置文件解析

然后在项目跟目录下面定义一个如下所示的 yaml 配置文件

hosts:
  - address: 192.168.1.1
    username: demo
    password: "demo"
    port: 22

  - address: 192.168.1.2
    username: demo
    password: "demo"
    port: 22

tasks:
  - name: list task
    command: ls -l /

我希望通过上面的配置文件来实现在 192.168.1.1 主机上查看 / 下面的内容,如果还有 更多主机或者任务,只需要在配置文件中添加即可,如果我们手动执行这些命令,需要登录到每台主机上执行,这样就会很麻烦,所以我们希望通过 python 来实现这个功能

要在程序中解析 YAML 文件,我们可以使用一个 PyYAML 库,可以通过 pip 安装:

$ pip3 install pyyaml

然后我们就可以在项目根目录下面建设一个名为 config.py 的文件,用于解析 YAML 配置文件,代码如下:

import yaml  # 导入 PyYAML 库


# 加载配置文件
def load_configuration(file_path: str):
    with open(file_path, 'r') as file:
        return yaml.safe_load(file)  # 解析 YAML 文件


if __name__ == '__main__':
    config = load_configuration('playbook.yaml')
    print(config)

上面代码中我们定义了一个 load 函数,用于加载 YAML 配置文件,要解析 YAML 文件很简单,只需要使用 yaml.safe_load 函数即可,它会将 YAML 文件解析为 Python 字典对象.

下面我们用 if name_ == '__main' 来测试一下,运行代码,这一段的意思是如果直接运行这个文件,就会执行 load 函数，加载配置文件并打印出来，运行结果如下：

$ python3 config.py
{'hosts': [{'address': '10.0.10.140', 'username': 'openbayes', 'password': 'openbayes', 'port': 22}], 'tasks': [{'name': 'list task', 'command': 'ls -l /'}]}

可以看到我们成功加载了配置文件,接下来我们就可以通过 SSH 连接到远程服务器并执行命令

远程执行任务

上面我们已经成功加载了配置文件,接下来我们就可以通过 ssh 连接到远程服务器并执行命令.要实现 SSH 连接,我们可以使用 paramiko 这个库实现,这个库是一个纯 Python 实现的 SSH 客户端,主要用于远程执行命令和文件传输等任务,它是一个功能强大的工具,适用与需要进行 SSH 连接的自动化任务,可以通过 pip 安装这个包;

$ pip3 install paramiko

下面我们先了解下 paramiko 的基本用法,然后再来实现远程执行任务的功能.

要连接到远程服务器,我们首先需要一个 SSHClient 对象,然后使用 connect 方法连接到远程服务器,示例代码如下:

import paramiko

# 创建 SSH client 对象
client = paramiko.SSHClient()

# 自动添加主机密钥,(如果之前没有连接过)
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

# 连接到远程服务器
client.connect(hostname='10.0.10.140', username='openbayes', password='openbayes')

# 执行命令
_, stdout, stderr = client.exec_command('hostname && hostname -I')

# 输出结果
print(stdout.read().decode())
print(stderr.read().decode())

# 关闭连接
client.close()
上面的代码中,我们首先创建了一个 SSHClient 对象,然后使用 connect 方法连接到远程服务器,接着使用 exec_command 方法执行命令,最后读取输出并关闭连接,这个整个过程其实就是一个典型的 SSH 连接过程

这里需要注意的是 exec_command 方法返回的是一个三元组,分别是标准输入,标准输出,和标准错误,我们可以通过这三个对象来读取命令的输入和输出.在读取输出的时候,需要使用 decode 方法将字节流转换为字符串.

将上面的代码保存到一个名为 ssh.py 的文件中,然后运行,可以看到输出结果:

可以看到我们成功连接到远程服务器并执行了 ls -l 命令

接下来我们可以将上面的代码封装成一个函数,用于执行远程命令,代码如下:

import paramiko

# 远程执行命令
def execute_remote_commond(host, port, command, username, password):
  try:
    print(f'Executing command: {command} on {host}')
    # 创建 SSHClient 对象
    client = paramiko.SSHClient()
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    # 连接服务器
    client.connect(host, port, username, password)
    # 执行命令
    _, stdout, stderr = client.exec_command(command)

    # 输出结果
    output = stdout.read().decode()
    error = stdout.read().decode()

    # 关闭连接
    client.close()
    return output, error
  except Exception as e:
    return None, str(e)

并发执行任务

上面我们已经实现了远程执行任务的功能,接下来我们需要实现并发执行任务的功能,即同时连接到多个远程服务器并执行命令,要实现并发执行任务,我们可以使用多线程或多进程来实现,由于我们这里是通过网络 IO 来实现并发,所以使用多线程更适合

前面我们学过 Python 提供了一个 concurrent.futures 模块,可以用来实现并发编程,其中有一个ThreadPoolExecutor 类,可以用来创建一个线程池,方便并发执行任务,我们可以使用这个类并发执行无误的功能

回到页面顶部