A Python library and command-line tool for streamlining SSH usage in application deployment and systems administration.
—
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
Essential remote and local command execution functions that form the foundation of Fabric's automation capabilities. These operations provide the primary interface for executing shell commands both locally and on remote hosts, with comprehensive support for privileged operations, output capture, and error handling.
Execute shell commands on remote hosts with full control over execution environment, output capture, and error handling.
def run(command, shell=True, pty=True, combine_stderr=None, quiet=False, warn_only=False, stdout=None, stderr=None, timeout=None, shell_escape=None):
"""
Execute shell command on remote host.
Args:
command (str): Shell command to execute
shell (bool): Execute via shell (default: True)
pty (bool): Request pseudo-terminal (default: True)
combine_stderr (bool): Combine stderr with stdout
quiet (bool): Suppress all output (default: False)
warn_only (bool): Continue on failure (default: False)
stdout (file): File-like object for stdout capture
stderr (file): File-like object for stderr capture
timeout (int): Command timeout in seconds
shell_escape (bool): Escape shell characters
Returns:
_AttributeString: Command output with execution metadata
.failed (bool): True if command failed
.succeeded (bool): True if command succeeded
.return_code (int): Command exit code
.command (str): Original command string
.real_command (str): Actual executed command
.stderr (str): Standard error output
"""Usage Examples:
from fabric.api import run, env
# Basic command execution
result = run('ls -la')
print(result) # Command output
print(result.return_code) # Exit code
# Execute with error handling
result = run('some-command', warn_only=True)
if result.failed:
print(f"Command failed with code {result.return_code}")
# Capture output to variables
output = run('cat /proc/meminfo', quiet=True)
if 'MemTotal' in output:
print("Found memory information")
# Set timeout for long-running commands
run('backup-database.sh', timeout=3600) # 1 hour timeoutExecute commands with superuser privileges using sudo, with support for custom users, groups, and sudo configuration.
def sudo(command, shell=True, pty=True, combine_stderr=None, user=None, quiet=False, warn_only=False, stdout=None, stderr=None, group=None, timeout=None, shell_escape=None):
"""
Execute command with superuser privileges.
Args:
command (str): Shell command to execute
shell (bool): Execute via shell (default: True)
pty (bool): Request pseudo-terminal (default: True)
combine_stderr (bool): Combine stderr with stdout
user (str): Target user for command execution (default: root)
quiet (bool): Suppress all output (default: False)
warn_only (bool): Continue on failure (default: False)
stdout (file): File-like object for stdout capture
stderr (file): File-like object for stderr capture
group (str): Target group for command execution
timeout (int): Command timeout in seconds
shell_escape (bool): Escape shell characters
Returns:
_AttributeString: Command output with execution metadata
"""Usage Examples:
from fabric.api import sudo, env
# Basic privileged command
sudo('systemctl restart nginx')
# Execute as specific user
sudo('npm install', user='nodeapp')
# Execute with specific group
sudo('chgrp -R webdev /var/www/', group='webdev')
# Handle sudo password prompts
env.sudo_password = 'mypassword'
sudo('apt-get update')
# Privileged file operations
sudo('cp config.json /etc/myapp/')
sudo('chmod 644 /etc/myapp/config.json')Execute commands on the local system where Fabric is running, useful for preprocessing, git operations, and local file management.
def local(command, capture=False, shell=None):
"""
Execute command on local system.
Args:
command (str): Shell command to execute
capture (bool): Capture output instead of printing (default: False)
shell (str): Shell to use for execution
Returns:
_AttributeString: Command output with execution metadata
"""Usage Examples:
from fabric.api import local, run
# Basic local command
local('git pull origin master')
# Capture local command output
git_status = local('git status --porcelain', capture=True)
if git_status:
print("Local changes detected")
# Local preprocessing before remote deployment
local('npm run build')
run('systemctl reload nginx') # Deploy on remote
# Use custom shell
local('echo $BASH_VERSION', shell='/bin/bash')Operations that provide user interaction capabilities during task execution.
def prompt(text, key=None, default='', validate=None):
"""
Prompt user for input with optional validation.
Args:
text (str): Prompt message to display
key (str): Environment key to store result
default (str): Default value if no input provided
validate (str|callable): Validation pattern or function
Returns:
str: User input value
"""
def reboot(wait=120, command='reboot'):
"""
Reboot remote system and wait for reconnection.
Args:
wait (int): Seconds to wait before reconnecting (default: 120)
command (str): Reboot command to execute (default: 'reboot')
"""
def open_shell(command=None):
"""
Open interactive shell on remote host.
Args:
command (str): Optional command to execute first
"""Usage Examples:
from fabric.api import prompt, reboot, open_shell, env
# Get user input
env.branch = prompt('Deploy which branch?', default='master',
validate=r'^[a-zA-Z0-9_-]+$')
# Confirm dangerous operations
if prompt('Delete database? (yes/no)', default='no') == 'yes':
sudo('dropdb myapp')
# Reboot and wait for system to come back online
reboot(wait=180) # Wait 3 minutes
# Drop into interactive shell for debugging
open_shell()
# Open shell with initial command
open_shell('cd /var/log && tail -f nginx/error.log')Verify that required environment variables and settings are configured before task execution.
def require(*keys, **kwargs):
"""
Check for required environment variables.
Args:
*keys: Environment keys that must be set
**kwargs: Environment keys with required values
Raises:
SystemExit: If required variables are missing
"""Usage Examples:
from fabric.api import require, env, task
@task
def deploy():
# Ensure required settings are configured
require('hosts', 'user', 'key_filename')
# Require specific values
require(warn_only=False, parallel=True)
# Continue with deployment
run('git pull')
# Set required environment
env.hosts = ['server1.example.com', 'server2.example.com']
env.user = 'deploy'
env.key_filename = '/path/to/deploy_key'All core operations return _AttributeString objects that provide rich metadata about command execution:
result = run('ls /nonexistent')
# Check execution status
if result.failed:
print(f"Command failed with exit code: {result.return_code}")
print(f"Error output: {result.stderr}")
if result.succeeded:
print("Command completed successfully")
# Access command information
print(f"Original command: {result.command}")
print(f"Actual executed command: {result.real_command}")
# Use as regular string
for line in result.splitlines():
print(f"Output line: {line}")Control what gets displayed during command execution:
from fabric.api import run, hide, show, settings
# Hide all output
with hide('everything'):
result = run('long-running-command')
# Show only specific output types
with show('stdout'):
run('echo "This will be shown"')
# Combine with other settings
with settings(hide('stderr'), warn_only=True):
run('command-that-might-fail')Install with Tessl CLI
npx tessl i tessl/pypi-fabric