Python中fabric库的实用函数分享与踩坑经验

最近一直在做的一个东西, 很多地方需要用到 shell 命令,
同样的命令既要在本地运行, 也要放到远程,并获取远程的输出结果
如果每个地方都要判断是在本地还是远程然后再去分别实例不同的对象,
所以写了一个函数, 尽量减少一下重复代码

def cmd_method(host, port=22, user=None, pwd=None):
    """
    执行远程或本地命令
    r = r"(?i)y\|n|\[y/d/n\]|\[y/n\]|y/n"
    watcher = Responder(pattern=r, response='y\n')
    Responder 对象可以通过正则匹配对 stdout 做分析, 如果匹配到了,就向 stdin 写入 response
    完成命令的反复交互
    :param host: 主机 ip, 本地可以用 'localhost' or '127.0.0.1'
    :param port: ssh 的端口
    :param user: 登陆远程主机的用户
    :param pwd: 远程登陆的密码
    :return: 正常输出和错误输出: (stdout, stderr)
    """

from fabric import Connection
from invoke import run, Responder
from paramiko import AuthenticationException
from paramiko.ssh_exception import NoValidConnectionsError, SSHException

def local(command, interactive):
    watcher = Responder(
            pattern=interactive['pattern'],
            response=interactive['response']) if interactive else None
    res = run(command, watchers=[watcher], warn=True, hide=True)
    res = res.stdout.encode('utf8'), res.stderr.encode('utf8')
    return res

def remote(command, interactive):
    if not ip_check(host):
        return 'host ip error'
    with Connection(host=host, port=port, user=user,
                    connect_kwargs={'password': pwd},
                    connect_timeout=10) as c:
        watcher = Responder(
                pattern=interactive['pattern'],
                response=interactive['response']) if interactive else []
        res = c.run(command, watchers=[watcher], warn=True, hide=True)
        res = res.stdout.encode('utf8'), res.stderr.encode('utf8')
        return res
if host and host in ('localhost' or '127.0.0.1'):
    return local
elif host and port and user and pwd:
    with Connection(host=host, port=port, user=user,
                    connect_kwargs={'password': pwd},
                    connect_timeout=10) as c:
        try:
            # 测试参数可用
            c.run('hostname', hide=True)
        except (AuthenticationException, NoValidConnectionsError,
                SSHException) as e:
            return e.__str__()
    return remote
else:
    return 'args error'

因为 fabric 的 Connection 的 run 方法也是继承自 invoke, 所以参数作用基本都是一样的
我最常用的是 warn 和 hide 还有 watchers
warn 默认为 False, 默认情况下会因为 shell 命令的错误输出而抛错, 也就是直接抛出 stderr
如果设为 True, 就会将 shell 命令的错误输出写到 Result 对象的 stderr 内

hide 也是默认为 False, 默认情况下将远程的输出信息在当前命令行输出, 为 True 时, 则不会, 但不论是什么, 都不会影响 Result 对象的 stdout 和 stderr 结果, 还可以只隐藏 stdout 或 stderr

watchers 参数, 传入的是一个包含诺干 Responder 实例的列表
当需要运行交互式的命令时, 可以用 Responder 对象来匹配输出, 并写入输入, 做自动化部署时很实用

还有一个 pty 参数, 这个参数, 默认是设为 True, 找文档时候发现很多人都是设为 True, 但在我踩过很多坑后,
我发现当设为 True 时, 有时标准输出(stdout)和错误输出(stderr)会混乱, 不方便后面的逻辑判断, 所以最好别动

还有 out_stream 和 err_stream, 可以将输出导到一个 write 模式打开的类 file 对象, 方便做记录


Python中fabric库的实用函数分享与踩坑经验

1 回复

Fabric是个好东西，但用起来确实有不少坑。分享几个我常用的实用函数和踩坑点。

1. 连接管理函数 别用execute直接跑命令，连接池管理不好容易乱。我习惯这么写：

from fabric import Connection, Config
from invoke import Responder

def get_connection(host, user, key_path):
    """带重试和超时的连接工厂"""
    config = Config(overrides={'connect_timeout': 10})
    conn = Connection(host=host, user=user, connect_kwargs={
        'key_filename': key_path
    }, config=config)
    
    # 测试连接
    try:
        conn.run('echo test', hide=True, warn=True)
        return conn
    except Exception as e:
        conn.close()
        raise ConnectionError(f"连接失败: {host} - {str(e)}")

2. 带交互的命令执行 处理sudo密码或者确认提示：

def run_with_sudo(conn, command, password=None):
    """处理sudo交互"""
    if password:
        sudopass = Responder(
            pattern=r'\[sudo\] password',
            response=password + '\n'
        )
        result = conn.run(
            f'sudo {command}',
            pty=True,
            watchers=[sudopass]
        )
    else:
        result = conn.run(f'sudo {command}', pty=True)
    
    if result.failed:
        raise RuntimeError(f"命令执行失败: {command}")
    return result.stdout.strip()

3. 文件传输的坑 put和get的路径处理要小心：

def safe_put(conn, local_path, remote_path=None):
    """安全的文件传输，自动处理路径"""
    if remote_path is None:
        remote_path = f'~/tmp/{os.path.basename(local_path)}'
    
    # 确保远程目录存在
    conn.run(f'mkdir -p {os.path.dirname(remote_path)}', hide=True)
    
    # 传输文件
    result = conn.put(local_path, remote_path)
    return result.remote

踩坑经验：

  1. 连接泄漏：用完一定要conn.close()，特别是长时间运行的任务
  2. 编码问题：远程输出默认是bytes，用.stdout获取字符串
  3. 超时设置：网络不好的时候，connect_timeout和command_timeout都要配
  4. pty模式：交互式命令必须加pty=True，否则会卡住
  5. 环境变量：远程环境可能和本地不同，命令要用绝对路径

建议：用连接池管理多个主机。

回到顶部