Mastering Fabric 2: Remote Automation with Python

Fabric

Fabric is a higher‑level wrapper built on Paramiko that simplifies remote command execution. It has three major versions: fabric1, fabric2, and the unofficial fabric3 (not recommended). The official documentation is available at www.fabfile.org.

Installation

pip install fabric

or pip install fabric2 installs the latest official version.

Usage

Connection

from fabric import Connection

conn = Connection(f"{user}@{host}:{port}",
                 connect_kwargs={"password": password},
                 connect_timeout=5)  # 5‑second timeout
# Multiple commands can be chained with && or ;
conn.run("ls")

Parameters for the run() method are:

Parameter

Description

hide=True

Suppresses server output from being printed to the console.

warn=True

Ignores non‑zero exit codes; errors are sent to stderr. If False, a SystemExit exception is raised.

pty=True

Enables an interactive pseudo‑terminal; other values are not recommended.

watchers

List of Responder objects for automatic interaction.

out_stream

File‑like object to capture standard output (similar to stdout in Fabric 1).

err_stream

File‑like object to capture error output.

The run() call returns an object whose attributes can be accessed as follows:

ret.stdout.strip()   # normal output
ret.stderr.strip()   # error output
ret.failed           # True if the command failed, otherwise False

Executing Interactive Commands

When a command requires user input, use Responder. The first argument is the pattern to match, the second is the response. Interactive commands must be run with pty=True, otherwise you will see errors such as “no tty present and no askpass program specified”. If sudo prompts appear, you need to run as root or modify the sudoers file.

from invoke import Responder
from fabric import Connection

sudopass = Responder(pattern=fr'\[sudo\] password for {user}:',
                      response=f'{password}
')
conn = Connection(f"{user}@{host}:{port}", connect_kwargs={"password": password})
conn.run("sudo whoami", pty=True, watchers=[sudopass])

When using Responder, ensure the matching string is escaped and the response ends with a newline ( \n) to simulate pressing Enter.

Executing Local Commands

# Using Connection.local (a wrapper around invoke.run)
conn.local("dir")

# Or directly with invoke.run
from invoke import run

Operating Multiple Machines

from fabric import SerialGroup as Group

results = Group('web1', 'web2', 'mac1').run('uname -s')
print(results)
for connection, result in results.items():
    print(f"{connection.host}: {result.stdout}")

# Upload and unpack example

def upload_and_unpack(c):
    if c.run('test -f /opt/mydata/myfile', warn=True).failed:
        c.put('myfiles.tgz', '/opt/mydata')
        c.run('tar -C /opt/mydata -xzvf /opt/mydata/myfiles.tgz')

for connection in Group('host1', 'host2', 'host3', user='root', connect_kwargs={'password': '123456'}):
    upload_and_unpack(connection)

Upload Files

Fabric uploads a single file at a time. To upload multiple files, either archive them locally and extract remotely, or iterate over a directory using pathlib.Path.rglob. Note that uploading to a non‑existent remote directory will raise an error.

conn = Connection(f"{user}@{host}:{port}", connect_kwargs={"password": password})
sftp = conn.sftp()

def put(local_path, remote_path):
    local_path, remote_path = Path(local_path), Path(remote_path)
    if local_path.is_dir():
        for path in local_path.rglob('[!.]*'):
            remote = remote_path.joinpath(path.relative_to(local_path))
            if path.is_file():
                check_remote_path(remote.parent, is_mkdir=True)
                conn.put(str(path), str(remote))
    else:
        check_remote_path(remote_path.parent, is_mkdir=True)
        if remote_isdir(remote_path):
            remote_path = remote_path.joinpath(local_path.name)
        conn.put(str(local_path), str(remote_path))

Download Files

Fabric’s download behavior differs slightly from Paramiko: it automatically creates the local directory if it does not exist, and if the local path does not include a filename, the remote filename is used as the default, provided the local path ends with a slash.

def traverse_remote_files(remote_path: Path, local_path: Path):
    files_attr = sftp.listdir_attr(str(remote_path))
    for file_attr in files_attr:
        filename = file_attr.filename
        if filename.startswith('.'):  # skip hidden files
            continue
        local = local_path.joinpath(filename)
        remote = remote_path.joinpath(filename)
        if stat.S_ISDIR(file_attr.st_mode):
            yield from traverse_remote_files(remote, local)
        else:
            yield remote

def remote_isdir(remote_path):
    attr = sftp.lstat(str(remote_path))
    return stat.S_ISDIR(attr.st_mode)

def get(remote_path, local_path):
    local_path, remote_path = Path(local_path), Path(remote_path)
    if remote_isdir(remote_path):
        for filepath in traverse_remote_files(remote_path, local_path):
            relpath = filepath.relative_to(remote_path)
            local = local_path.joinpath(relpath)
            try:
                conn.get(str(filepath), str(local))
            except FileNotFoundError:
                print("File not found")
    else:
        if not local_path.suffix:
            local_path = local_path.joinpath(remote_path.name)
        try:
            conn.get(str(remote_path), str(local_path))
        except FileNotFoundError:
            print("File not found")

fab Command

Run fab --help to see available commands. Below are common fab 2 options, which differ significantly from fabric 1.

Option

Description

-l

List tasks (functions decorated with @task).

-c

Load a specific task module (default is fabfile.py).

-r

Search for task modules starting from a root directory.

-f

Specify a configuration file path.

-H

Specify target hosts, separated by commas.

-V

Show versions of Fabric, Paramiko, and Invoke.

-w

Same as warn=True; prevents exceptions from stopping execution.

Example task definitions:

@task
def task1(c, param):
    print(f"hello world, today is {param}")

@task
def task2(c):
    ...
# Execute with: fab -H localhost task2

Functions decorated with @task must accept a Context argument; otherwise a TypeError is raised. If the entry file is not named fabfile.py, use -c to specify the collection module.

"""Call a remote script from the local machine"""
now = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
conn.run(f"fab -c /home/yuqiuwen/PythonProjects/zhongxin/fab task1 --param '{now}'")

Listing tasks in JSON format:

fab -r ./ -c fab -l -F json
# Output: {"name": "fab", "tasks": [{"name": "task1"}, {"name": "task2"}], ...}

Encapsulation

from fabric import Connection
from pathlib import Path
import stat
import logging

class FabConnection:
    def __init__(self, host, user, password, port=22, mylogger=None):
        self._host = host
        self._user = user
        self._password = password
        self._port = port
        self._sftp = None
        self._conn = None
        self.mylogger = mylogger
        self.connect()

    def connect(self):
        conn = Connection(f"{self._user}@{self._host}:{self._port}",
                         connect_kwargs={"password": self._password})
        self._conn = conn
        self._sftp = conn.sftp()

    def get(self, remote_path, local_path):
        local_path, remote_path = Path(local_path), Path(remote_path)
        if self.remote_isdir(remote_path):
            for filepath in self.traverse_remote_files(remote_path, local_path):
                relpath = filepath.relative_to(remote_path)
                local = local_path.joinpath(relpath)
                try:
                    self._conn.get(self.normpath(filepath), self.normpath(local))
                    self.writer(f"download {relpath} successful!")
                except FileNotFoundError:
                    self.writer(f"FileNotFoundError: {filepath} -> {local}", level='error')
        else:
            if not local_path.suffix:
                local_path = local_path.joinpath(remote_path.name)
            try:
                self._conn.get(self.normpath(remote_path), self.normpath(local_path))
                self.writer(f"download {remote_path} successful!")
            except FileNotFoundError:
                self.writer(f"FileNotFoundError: {remote_path} -> {local_path}", level='error')

    def put(self, local_path, remote_path):
        local_path, remote_path = Path(local_path), Path(remote_path)
        if local_path.is_dir():
            for path in local_path.rglob('[!.]*'):
                remote = remote_path.joinpath(path.relative_to(local_path))
                if path.is_file():
                    self.check_remote_path(remote.parent, is_mkdir=True)
                    self._conn.put(self.normpath(path), self.normpath(remote))
        else:
            self.check_remote_path(remote_path.parent, is_mkdir=True)
            if self.remote_isdir(remote_path):
                remote_path = remote_path.joinpath(local_path.name)
            self._conn.put(self.normpath(local_path), self.normpath(remote_path))

    def traverse_remote_files(self, remote_path: Path, local_path: Path):
        files_attr = self._sftp.listdir_attr(self.normpath(remote_path))
        for file_attr in files_attr:
            filename = file_attr.filename
            if filename.startswith('.'):  # skip hidden files
                continue
            local = local_path.joinpath(filename)
            remote = remote_path.joinpath(filename)
            if stat.S_ISDIR(file_attr.st_mode):
                yield from self.traverse_remote_files(remote, local)
            else:
                yield remote

    def close(self):
        if self._conn:
            self._conn.close()
        if self._sftp:
            self._sftp.close()
        self._conn, self._sftp = None, None

    def writer(self, message, level=None):
        if self.mylogger:
            if isinstance(self.mylogger, logging.Logger):
                if not level:
                    self.mylogger.info(message)
                elif level == 'error':
                    self.mylogger.error(message)
                else:
                    self.mylogger.warning(message)
            else:
                self.mylogger.write(message + "
")
        else:
            print(message + "
")

    @staticmethod
    def normpath(path):
        if isinstance(path, Path):
            path = str(path)
        return path.replace('\\', '/')

    def remote_isdir(self, remote_path):
        attr = self._sftp.lstat(self.normpath(remote_path))
        return stat.S_ISDIR(attr.st_mode)

    def check_remote_path(self, remote_path, is_mkdir=False):
        remote_path = self.normpath(remote_path)
        try:
            self._sftp.lstat(remote_path)
            return True
        except FileNotFoundError:
            if is_mkdir:
                try:
                    self._sftp.mkdir(remote_path)
                except PermissionError:
                    ret = self.run(f"sudo mkdir {remote_path}", pty=True, watchers=[self.sudo_pass])
                    return not ret[2]
                return True
            else:
                return False

    @property
    def sudo_pass(self):
        sudopass = Responder(pattern=fr'[sudo] password for {self._user}:',
                             response=f'{self._password}
')
        return sudopass

    def run(self, cmd, hide=False, warn=True, pty=False, watchers=None):
        ret = self._conn.run(cmd, hide=hide, warn=warn, pty=pty,
                             out_stream=self.mylogger, err_stream=self.mylogger,
                             watchers=watchers, encoding="utf8")
        return ret.stdout.strip(), ret.stderr.strip(), ret.failed

    @property
    def conn(self):
        return self._conn

# Demo usage
demo = FabConnection(host='172.16.101.222', user='xj', password='king')
demo.get('/home/xj/Desktop/dir_a', './')
demo.put('./', '/home/xj/Desktop/dir_cc')
demo.run('ls -l')