Fabric模块是一个SSH 协议的Python 库，基于Paramiko封装，核心功能是

远程执行命令和文件传输，尤其擅长批量操作多台服务器，它的API更简洁，非常适合 “测试环境批量部署、多节点运维操作和检查” 等场景。

官方文档：Welcome to Fabric’s documentation! — Fabric documentation

核心优势

批量操作

：同时管理数十台甚至上百台服务器，命令一键下发；

简洁语法

：比直接使用 Paramiko 更简单，几行代码实现复杂操作；

丰富功能

：支持命令执行、文件上传下载、sudo 权限、结果校验等；

可扩展性

：结合 Python 语法，轻松编写复杂的自动化流程。

环境搭建

Fabric 目前有两个主流版本：Fabric1.x（旧版，基于函数）和 Fabric2.x（新版，基于对象）。推荐使用 2.x 版本，语法更现代：

# 安装Fabric最新的2.x版本pip install fabric==2.7.1  -i https://mirrors.aliyun.com/pypi/simple/#指定版本pip install Fabric -i https://mirrors.aliyun.com/pypi/simple/#最新版本

导入方式：

from fabric import Connection, Configfrom fabric.group import SerialGroup # 用于单线程串行ch批量管理多台服务器from fabric.group import ThreadingGroup  # 用于多线程并行批量管理多台服务器

一、核心功能详解：连接服务器与执行命令

1. 连接服务器：fabric.Connection()

功能：创建与单台服务器的SSH 连接，是所有操作的基础。

参数说明

案例 1：密码连接服务器

from fabric import Connection# 创建连接（host格式：用户名@IP:端口）conn = Connection(    host="test@192.168.1.100",    connect_kwargs={"password": "test123"}  # 密码认证)# 测试连接（执行简单命令）result = conn.run("uname -a", hide=True)  # hide=True 隐藏命令输出print(f"服务器系统信息：{result.stdout.strip()}")#运行后记得关闭连接conn.close()

Connection类支持上下文管理器，因此可以使用with语句，防止会忘掉使用close关闭连接

以上代码可以更新为：

from fabric import Connection# 创建连接（host格式：用户名@IP:端口）conn = Connection(    host="test@192.168.1.100",    connect_kwargs={"password": "test123"}  # 密码认证)with conn as f: # 测试连接（执行简单命令）  result = f.run("uname -a", hide=True)  # hide=True 隐藏命令输出  print(f"服务器系统信息：{result.stdout.strip()}")

案例 2：密钥连接服务器（更安全）

from fabric import Connection# 密钥连接（私钥路径）conn = Connection(    host="test@192.168.1.101",    connect_kwargs={"key_filename": "~/.ssh/id_rsa",  # 本地私钥路径"passphrase": "key123"# 若私钥有密码，需指定（无则省略）    })with conn as f: # 检查服务器磁盘使用率  result = conn.run("df -h /", hide=True)  print(f"根目录磁盘使用情况：\n{result.stdout}")

2. 执行命令：Connection.run()（重点函数）

功能：在远程服务器上执行Shell命令，支持输出控制、结果校验、超时设置等。

参数说明

返回值

Result 对象，包含命令执行结果：

• stdout：命令输出内容（字符串）；
• stderr：错误输出内容；
• exit_code：退出状态码（0表示成功）；
• ok：布尔值，True表示命令成功执行。

案例：执行命令并处理结果

from fabric import Connectionconn = Connection(    host="test@192.168.1.100",    connect_kwargs={"password": "test123"})try:# 执行命令（查看被测应用进程）    result = conn.run("ps -ef | grep test_app | grep -v grep",        hide=False,  # 显示命令输出        timeout=10# 超时时间10秒    )# 处理结果    if result.ok:        print(f"✅ 应用进程存在：\n{result.stdout}")else:        print(f"❌ 命令执行失败，状态码：{result.exit_code}")except Exception as e:    print(f"❌ 执行命令出错：{str(e)}")finally:	 conn.close()

3. 执行sudo命令：Connection.sudo()

功能：以管理员权限执行命令（类似 sudo cmd），自动处理密码输入。

参数说明

案例：重启服务（需要sudo权限）

from fabric import Connectionconn = Connection(    host="test@192.168.1.100",    connect_kwargs={"password": "test123"})# 执行 sudo 命令（重启测试应用）result = conn.sudo("systemctl restart test_app",    password="test123",  # sudo 密码（与登录密码一致时可省略）    hide=False)if result.ok:    print("应用服务已重启")

4. 单线程批量管理多台服务器：SerialGroup与GroupResult

功能：通过Group创建多服务器连接组，批量执行命令，是Fabric的核心优势。

SerialGroup初始化时需要传入hosts信息的可迭代对象，以及登录的密码等信息，也支持通过with语句创建上下文，并且它的函数跟Connection类似，也有run()，get()，put()，close()等函数，返回的是GroupResult对象（对象是以每个Connection为key，以Result对象为value）。

案例：批量检查多台服务器的应用状态（每台服务器密码一样）

from fabric import Connectionfrom fabric.group import SerialGroup# 定义多台服务器（IP列表）servers = ["test@192.168.1.100","test@192.168.1.101","test@192.168.1.102"]# 创建服务器组（统一设置连接参数,对组内每个服务都使用Connection连接，生成的多个Connection实例）conn_group = SerialGroup(    *servers,    connect_kwargs={"password": "test123"}  # 所有服务器共用密码)# 批量执行命令（检查test_app进程数）results = conn_group.run("ps -ef | grep test_app | grep -v grep | wc -l",    hide=True,    warn=True# 某台服务器失败不影响其他)# 解析批量执行结果print("=== 批量检查应用进程结果 ===")for conn, result in results.items():    if result.ok:        process_count = result.stdout.strip()        status = "运行中"if int(process_count) > 0else"已停止"        print(f"{conn.host}：{status}（进程数：{process_count}）")    else:        print(f"{conn.host}：执行失败（{result.stderr.strip()}）")#关闭连接conn_group.close()

输出示例

=== 批量检查应用进程结果 ===test@192.168.1.100：运行中（进程数：2）test@192.168.1.101：已停止（进程数：0）test@192.168.1.102：执行失败（bash: ps: command not found）

三、文件传输：上传与下载（重点功能）

1. 上传文件：Connection.put()

功能：将本地文件 / 目录上传到远程服务器。

参数说明

案例：上传测试脚本到服务器

from fabric import Connectionconn = Connection(    host="test@192.168.1.100",    connect_kwargs={"password": "test123"})# 上传本地测试脚本到服务器 /opt/scripts/ 目录result = conn.put(    local="./run_test.py",    remote="/opt/scripts/run_test.py",    confirm=True# 验证上传完整性)if result.failed:    print(f"❌ 上传失败：{result.stderr}")else:    print(f"✅ 上传成功：{result.local} → {result.remote}")# 上传后赋予执行权限    conn.run("chmod +x /opt/scripts/run_test.py")

2. 下载文件：Connection.get()

功能：将远程服务器文件 / 目录下载到本地。

参数说明

案例：批量下载多台服务器的日志

from fabric import Connectionfrom fabric.group import SerialGroupimport os# 服务器列表servers = ["test@192.168.1.100", "test@192.168.1.101"]conn_group = SerialGroup(*servers, connect_kwargs={"password": "test123"})# 本地日志目录（按服务器IP分文件夹）local_log_dir = "./server_logs"os.makedirs(local_log_dir, exist_ok=True)# 批量下载日志for conn in conn_group:# 提取服务器IP（从host字符串中解析）    server_ip = conn.host.split("@")[-1].split(":")[0]    local_path = f"{local_log_dir}/{server_ip}_test_app.log"    print(f"开始下载 {conn.host} 的日志...")    result = conn.get(        remote="/var/log/test_app.log",        local=local_path    )    if result.ok:        print(f"✅ 已保存到：{local_path}")    else:        print(f"❌ {conn.host} 日志下载失败")

四、ThreadingGroup：多线程批量管理SSH连接

ThreadingGroup与SerialGroup（串行执行）不同， 用于对多个远程主机并行执行任务，会同时在所有主机上发起命令，无需等待前一个主机完成，从而节省时间。适合需要提高多主机处理效率的场景（例如批量部署、状态检查依赖关系的任务）。

ThreadingGroup的接口与SerialGroup类似，接收主机列表和配置参数，通过with语句创建上下文，调用 run()、sudo() 等方法时会并行执行命令。

五、配置全局参数：Config

Config类用于集中管理Fabric的配置项，包括连接配置（如主机、用户、端口）、运行时选项（如超时时间、并行执行，sudo密码、输出格式等）等，避免重复配置。

示例 1：基础配置与连接

from fabric import Config, Connection# 创建自定义配置custom_config = Config(    overrides={# 连接相关配置"user": "your_username",       # 默认用户名"connect_kwargs": {"password": "your_password",  # 密码（也可使用密钥）# 或使用 SSH 密钥# "key_filename": "/path/to/private/key",        },# 执行相关配置"run": {"warn": True,  # 命令执行失败时不抛出异常，仅返回结果"echo": True,  # 打印执行的命令        }    })# 使用自定义配置创建连接c = Connection(    host="remote_host_ip",  # 远程主机地址    config=custom_config    # 传入配置)# 执行远程命令result = c.run("ls -l")print(result.stdout)  # 输出命令结果

示例 2：配置超时与并行执行

from fabric import Config, SerialGroup# 配置超时时间和并行执行config = Config(    overrides={"connect_kwargs": {"timeout": 10,  # 连接超时（秒）        },"run": {"timeout": 30,  # 命令执行超时（秒）        }        }    })# 多主机批量操作（使用配置中的并行执行）hosts = ["host1", "host2", "host3"]with SerialGroup(*hosts, config=config) as group:    group.run("uptime")  # 同时在多主机上执行 `uptime` 命令

示例 3：通过配置文件加载

除了代码中硬编码，还可以通过配置文件（如 fabric.yaml 或 ~/.fabric.yaml）加载配置，Config 会自动读取这些文件：

# fabric.yamluser:"default_user"connect_kwargs:key_filename:"~/.ssh/id_rsa"run:echo:true

在代码中直接使用默认配置（会自动读取上述文件）：

from fabric import Connection, Config# 自动加载默认配置文件（如 fabric.yaml）config = Config()c = Connection(host="remote_host", config=config)c.run("who am i")

3. 核心配置项说明

Config 的配置项通过overrides参数传入，常见键包括：

• user：默认 SSH 用户名。
• port：默认 SSH 端口（默认 22）。
• connect_kwargs：传递给 paramiko（SSH 底层库）的参数，如 password、key_filename、timeout等。
• run：与 c.run() 相关的配置，如 warn（是否忽略错误）、echo（是否打印命令）、pty（是否使用伪终端）等。
• sudo：与 c.sudo() 相关的配置，如 password（sudo 密码）、user（sudo 切换的用户）等。
• executor：任务执行器配置，如 parallel（是否并行执行多主机任务）。

六、避坑指南：Fabric常见问题解决

1. 连接超时或拒绝连接

2. sudo 命令提示 “密码错误”

3. 并行执行时输出混乱

4. 大文件上传 / 下载速度慢

5. 命令执行无响应（卡在 PTY 分配）

六、总结：Fabric最佳实践

1. 批量操作优先用 Group
   ：多台服务器管理用 Grou替代循环单个 Connection，配合 parallel=True 提升效率；
2. 全局配置简化代码
   ：通过Config统一设置密码、超时时间，避免重复参数；
3. 敏感信息不硬编码
   ：密码、密钥路径等敏感信息，通过配置文件或环境变量传入（如 os.getenv("SSH_PASSWORD")）。

七、工具选择建议：Paramiko 还是 Fabric？

掌握 Fabric 后，你会发现批量服务器管理变得异常简单：从日常巡检到批量部署，几行代码就能替代数小时的手动操作。无论是测试工程师、运维人员还是开发人员，Fabric 都是提升工作效率的必备工具。