Zum Inhalt springen
1337skills 1337skills - Referenzen

Fabric (Python) Cheat Sheet

Overview

Fabric is a Python library and command-line tool for streamlining SSH-based application deployment and system administration tasks. It provides a Pythonic interface for executing shell commands remotely, transferring files, and managing connections to multiple servers. Fabric 2.x is built on top of Invoke (for local tasks) and Paramiko (for SSH).

Fabric replaces shell scripts with Python functions that can run commands locally and remotely, handle file transfers, and manage complex deployment workflows. It is widely used for automating deployments, server provisioning, and operational tasks across fleets of servers.

Installation

pip install fabric

# With additional SSH features
pip install fabric[pytest]

# Verify
fab --version
python -c "import fabric; print(fabric.__version__)"

Core Usage

Basic fabfile.py

from fabric import Connection, task

@task
def deploy(c):
    """Deploy the application to production."""
    conn = Connection('web1.example.com')
    with conn.cd('/opt/myapp'):
        conn.run('git pull origin main')
        conn.run('pip install -r requirements.txt')
        conn.run('python manage.py migrate')
        conn.run('sudo systemctl restart myapp')

@task
def check(c):
    """Check server status."""
    conn = Connection('web1.example.com')
    conn.run('uptime')
    conn.run('df -h')
    conn.run('free -m')

@task
def logs(c):
    """View application logs."""
    conn = Connection('web1.example.com')
    conn.run('tail -100 /var/log/myapp/app.log')
# Run tasks
fab deploy
fab check
fab logs

Connection Management

from fabric import Connection

# Basic connection
conn = Connection('web1.example.com')

# With username
conn = Connection('deploy@web1.example.com')

# With specific port and key
conn = Connection(
    host='web1.example.com',
    user='deploy',
    port=2222,
    connect_kwargs={
        'key_filename': '/path/to/key.pem',
    }
)

# Using SSH config
conn = Connection('web1')  # Uses ~/.ssh/config

# With password (use keys instead when possible)
conn = Connection(
    host='web1.example.com',
    user='deploy',
    connect_kwargs={'password': 'secret'}
)

Commands and Operations

Running Commands

from fabric import Connection, task

@task
def server_info(c):
    conn = Connection('web1.example.com')

    # Run command and capture output
    result = conn.run('uname -a', hide=True)
    print(f"OS: {result.stdout.strip()}")

    # Check return code
    result = conn.run('test -f /etc/myapp.conf', warn=True)
    if result.ok:
        print("Config file exists")
    else:
        print("Config file missing")

    # Run with sudo
    conn.sudo('systemctl restart nginx')
    conn.sudo('apt update && apt upgrade -y')

    # Run with environment variables
    conn.run('echo $APP_ENV', env={'APP_ENV': 'production'})

    # Change directory
    with conn.cd('/opt/myapp'):
        conn.run('git status')
        conn.run('pip install -r requirements.txt')

File Transfers

@task
def upload_config(c):
    conn = Connection('web1.example.com')

    # Upload file
    conn.put('local/config.yaml', '/etc/myapp/config.yaml')

    # Download file
    conn.get('/var/log/myapp/app.log', 'local/app.log')

    # Upload with sudo (upload to temp, then move)
    conn.put('nginx.conf', '/tmp/nginx.conf')
    conn.sudo('mv /tmp/nginx.conf /etc/nginx/nginx.conf')
    conn.sudo('systemctl reload nginx')

Local Commands

from invoke import task

@task
def build(c):
    """Build locally before deploying."""
    c.run('npm run build')
    c.run('docker build -t myapp:latest .')
    c.run('docker push myregistry/myapp:latest')

Configuration

Multi-Server Deployment

from fabric import Connection, SerialGroup, ThreadingGroup, task

WEB_SERVERS = [
    'web1.example.com',
    'web2.example.com',
    'web3.example.com',
]

@task
def deploy(c):
    """Deploy to all web servers."""
    group = SerialGroup(*WEB_SERVERS, user='deploy')
    group.run('cd /opt/myapp && git pull')
    group.run('cd /opt/myapp && pip install -r requirements.txt')
    group.run('sudo systemctl restart myapp')

@task
def deploy_parallel(c):
    """Deploy to all servers in parallel."""
    group = ThreadingGroup(*WEB_SERVERS, user='deploy')
    group.run('cd /opt/myapp && git pull')

@task
def rolling_deploy(c):
    """Rolling deploy one server at a time."""
    for host in WEB_SERVERS:
        conn = Connection(host, user='deploy')
        conn.sudo('systemctl stop myapp')
        with conn.cd('/opt/myapp'):
            conn.run('git pull origin main')
            conn.run('pip install -r requirements.txt')
        conn.sudo('systemctl start myapp')
        # Wait for health check
        conn.run('sleep 10 && curl -f http://localhost:8080/health')
        print(f"Deployed to {host}")

SSH Config Integration

from fabric import Config, Connection

# Use SSH config file
config = Config(ssh_config=True)
conn = Connection('myserver', config=config)

# Custom fabric config
config = Config(overrides={
    'run': {
        'echo': True,
        'warn': True,
    },
    'sudo': {
        'password': 'mypassword',
    },
    'connect_kwargs': {
        'key_filename': '/path/to/key.pem',
    },
})

Advanced Usage

Complete Deployment Pipeline

from fabric import Connection, task
from invoke import Exit
import datetime

@task
def deploy(c, branch='main', skip_tests=False):
    """Full deployment pipeline."""
    conn = Connection('web1.example.com', user='deploy')

    if not skip_tests:
        print("Running tests locally...")
        result = c.run('pytest tests/', warn=True)
        if not result.ok:
            raise Exit("Tests failed! Aborting deploy.")

    timestamp = datetime.datetime.now().strftime('%Y%m%d_%H%M%S')

    with conn.cd('/opt/myapp'):
        # Create backup
        conn.run(f'cp -r current backup_{timestamp}')

        # Pull latest code
        conn.run(f'git fetch && git checkout {branch} && git pull')

        # Install dependencies
        conn.run('pip install -r requirements.txt')

        # Run migrations
        conn.run('python manage.py migrate --noinput')

        # Collect static files
        conn.run('python manage.py collectstatic --noinput')

        # Restart application
        conn.sudo('systemctl restart myapp')

        # Verify health
        result = conn.run('curl -sf http://localhost:8080/health', warn=True)
        if not result.ok:
            print("Health check failed! Rolling back...")
            conn.run(f'cp -r backup_{timestamp} current')
            conn.sudo('systemctl restart myapp')
            raise Exit("Deploy failed and was rolled back.")

    print(f"Deploy of {branch} complete!")

@task
def rollback(c, backup_name=None):
    """Rollback to a previous version."""
    conn = Connection('web1.example.com', user='deploy')
    if backup_name is None:
        result = conn.run('ls -t /opt/myapp/backup_* | head -1', hide=True)
        backup_name = result.stdout.strip()

    with conn.cd('/opt/myapp'):
        conn.run(f'cp -r {backup_name} current')
        conn.sudo('systemctl restart myapp')
    print(f"Rolled back to {backup_name}")

Watchers for Interactive Prompts

from invoke import Responder

@task
def setup_db(c):
    conn = Connection('db.example.com')
    sudopass = Responder(
        pattern=r'\[sudo\] password:',
        response='mypassword\n',
    )
    conn.run('sudo createdb myapp', watchers=[sudopass])

Troubleshooting

IssueSolution
SSH connection refusedCheck SSH key, port, and firewall rules
Permission deniedVerify user permissions; use sudo()
Command hangsSet timeout parameter; check for interactive prompts
File upload failsCheck remote directory permissions
Parallel execution errorsUse SerialGroup for debugging; check for race conditions
fab command not foundEnsure fabric is installed; check PATH
# Debug connection
fab -d deploy

# List available tasks
fab --list

# Run with specific host
fab -H web1.example.com deploy

# Verbose output
fab deploy --echo