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, Config
from fabric.group import SerialGroup # 用于单线程串行ch批量管理多台服务器
from fabric.group import ThreadingGroup  # 用于多线程并行批量管理多台服务器

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

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

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

参数说明

参数名	类型	是否必填	说明	示例
host	str	是	服务器地址，格式支持多种：“user@host“, “host:port“, ，“user@host:port`	“test@192.168.1.100:22”
user	str	否	登录用户名（若host中已包含则可省略）	“test”
port	int	否	SSH端口，默认22（若host中已包含则可省略）	22
connect_kwargs	dict	否	连接参数（密码、密钥等）	{“password”: “123456”} 或 {“key_filename”: “~/.ssh/id_rsa”}
config	Config	否	全局配置对象（如超时时间、sudo 密码）	Config(overrides={“sudo”: {“password”: “123456”}})

案例 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"根目录磁盘使用情况：
{result.stdout}")

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

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

参数说明

参数名	类型	是否必填	说明	示例
command	str	是	要执行的 Shell 命令	“ls -l /opt”
hide	bool/str	否	是否隐藏输出：False（默认，显示所有）、True（隐藏所有）、”stdout”（隐藏标准输出）	hide=”stderr”（只隐藏错误输出）
warn	bool	否	命令执行失败（返回非 0 状态码）时是否只警告不抛异常，默认 False（抛异常）	warn=True（允许命令失败）
env	dict	否	设置环境变量	{“PATH”: “/usr/local/bin:$PATH”}
timeout	int	否	命令执行超时时间（秒）	30（最多等 30 秒）
pty	bool	否	是否请求伪终端（某些命令如 sudo、top 需要）	True（用于交互式命令）

返回值

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

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

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

from fabric import Connection

conn = 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"✅ 应用进程存在：
{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），自动处理密码输入。

参数说明

参数名	类型	说明
基本参数与 run() 一致
password	str	sudo 密码（若与登录密码一致，可省略）

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

from fabric import Connection

conn = 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 Connection
from 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) > 0 else "已停止"
        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()

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

参数说明

参数名	类型	是否必填	说明	示例
local	str	是	本地文件 / 目录路径	“./test_script.py” 或 “./configs/”
remote	str	否	远程目标路径（默认与本地文件名一样）	“/opt/test_script.py”
preserve_mode	bool	否	是否保留本地文件权限，默认 True	False（不保留权限）
confirm	bool	否	上传后是否验证文件大小，默认 True	True（推荐开启）

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

from fabric import Connection

conn = 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()

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

参数说明

参数名	类型	是否必填	说明	示例
remote	str	是	远程文件 / 目录路径	“/var/log/test_app.log”
local	str	否	本地目标路径（默认当前目录）	“./logs/test_app.log”
preserve_mode	bool	否	是否保留远程文件权限，默认 True
confirm	bool	否	下载后是否验证文件大小，默认 True

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

from fabric import Connection
from fabric.group import SerialGroup
import 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.yaml
user: "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("whoami")

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. 连接超时或拒绝连接

• 检查：服务器 IP、端口是否正确，SSH 服务是否启动（systemctl status sshd）；
• 解决：测试环境关闭防火墙（systemctl stop firewalld），或开放 22 端口。

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

• 检查：sudo 密码是否正确，服务器是否允许该用户使用 sudo（/etc/sudoers 配置）；
• 解决：通过 config 全局设置 sudo 密码，或在 sudo() 方法中显式指定 password 参数。

3. 并行执行时输出混乱

• 缘由：多台服务器同时输出，导致控制台信息交错；
• 解决：用 hide=True 隐藏输出，仅在结果处理时打印；为每个服务器的输出添加前缀（如 f”[{host}] {result.stdout}”）。

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

• 解决：压缩文件后传输（如 tar.gz）；增加并行数（parallel=True）；禁用 confirm 验证（confirm=False，但可能导致文件不完整）。

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

• 缘由：某些命令需要交互（如 vi、top），或服务器禁用了伪终端；
• 解决：非交互命令设置 pty=False（默认）；交互命令必须设置 pty=True，并通过 stdin 输入（复杂场景提议用 paramiko）。

六、总结：Fabric最佳实践

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

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

场景需求	推荐工具	核心缘由
单台服务器精细操作（如自定义命令流）	Paramiko	底层灵活，支持细粒度控制
多台服务器批量操作（如批量部署、检查）	Fabric	高层封装，一行代码实现批量任务
复杂交互场景（如动态输入密码）	Paramiko	支持直接操作stdin流
快速落地简单任务（如上传文件、执行单条命令）	Fabric	API 简洁，学习成本低

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