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
File transfer capabilities and advanced file management functions for uploading, downloading, template rendering, text processing, and project synchronization. These operations provide comprehensive file handling capabilities for deployment and configuration management workflows.
Core functions for uploading and downloading files between local and remote systems with comprehensive options for permissions, ownership, and error handling.
def put(local_path=None, remote_path=None, use_sudo=False, mirror_local_mode=False, mode=None, use_glob=True, temp_dir=""):
"""
Upload files to remote host.
Args:
local_path (str): Local file/directory path to upload
remote_path (str): Remote destination path
use_sudo (bool): Use sudo for file operations (default: False)
mirror_local_mode (bool): Copy local file permissions (default: False)
mode (str|int): Set specific file permissions (octal or symbolic)
use_glob (bool): Allow glob patterns in local_path (default: True)
temp_dir (str): Temporary directory for staged uploads
Returns:
_AttributeList: List of uploaded file paths with .failed/.succeeded attributes
"""
def get(remote_path, local_path=None, use_sudo=False, temp_dir=""):
"""
Download files from remote host.
Args:
remote_path (str): Remote file/directory path to download
local_path (str): Local destination path (default: same as remote filename)
use_sudo (bool): Use sudo for file access (default: False)
temp_dir (str): Temporary directory for staged downloads
Returns:
_AttributeList: List of downloaded file paths with .failed/.succeeded attributes
"""Usage Examples:
from fabric.api import put, get, run, cd
# Basic file upload
put('local-config.json', '/etc/myapp/config.json')
# Upload with sudo and specific permissions
put('nginx.conf', '/etc/nginx/sites-available/myapp',
use_sudo=True, mode='644')
# Upload directory with glob patterns
put('dist/*', '/var/www/myapp/', use_glob=True)
# Mirror local permissions
put('script.sh', '/opt/scripts/deploy.sh',
mirror_local_mode=True, use_sudo=True)
# Basic file download
get('/var/log/nginx/error.log')
# Download to specific location
get('/etc/nginx/nginx.conf', 'backup/nginx.conf')
# Download with sudo for protected files
get('/var/log/auth.log', use_sudo=True)
# Check transfer results
results = put('app.tar.gz', '/tmp/')
if results.succeeded:
print(f"Uploaded {len(results)} files successfully")
for failed_file in results.failed:
print(f"Failed to upload: {failed_file}")Extended file operations from the fabric.contrib.files module providing file existence checking, text processing, and template rendering capabilities.
def exists(path, use_sudo=False, verbose=False):
"""
Check if remote path exists.
Args:
path (str): Remote path to check
use_sudo (bool): Use sudo for access check (default: False)
verbose (bool): Show command output (default: False)
Returns:
bool: True if path exists
"""
def is_link(path, use_sudo=False, verbose=False):
"""
Check if remote path is a symbolic link.
Args:
path (str): Remote path to check
use_sudo (bool): Use sudo for access (default: False)
verbose (bool): Show command output (default: False)
Returns:
bool: True if path is a symbolic link
"""
def first(*args, **kwargs):
"""
Return first existing path from arguments.
Args:
*args: Paths to check in order
**kwargs: Keyword arguments passed to exists()
Returns:
str: First existing path or None
"""Usage Examples:
from fabric.contrib.files import exists, is_link, first
from fabric.api import run, sudo
# Check if file exists before operations
if exists('/etc/nginx/nginx.conf'):
run('nginx -t') # Test configuration
else:
print("Nginx config not found")
# Check with sudo for protected paths
if exists('/var/log/secure', use_sudo=True):
sudo('tail /var/log/secure')
# Check if path is symbolic link
if is_link('/etc/nginx/sites-enabled/default'):
run('readlink /etc/nginx/sites-enabled/default')
# Find first existing configuration file
config_path = first(
'/etc/myapp/config.json',
'/opt/myapp/config.json',
'/usr/local/etc/myapp.conf'
)
if config_path:
print(f"Found config at: {config_path}")
# Conditional operations based on existence
if not exists('/var/www/myapp'):
sudo('mkdir -p /var/www/myapp')
sudo('chown www-data:www-data /var/www/myapp')Upload and render template files with variable substitution using Jinja2 or simple string formatting.
def upload_template(filename, destination, context=None, use_jinja=False, template_dir=None, use_sudo=False, backup=True, mirror_local_mode=False, mode=None, pty=None):
"""
Upload rendered template file to remote host.
Args:
filename (str): Template filename (relative to template_dir)
destination (str): Remote destination path
context (dict): Variables for template rendering
use_jinja (bool): Use Jinja2 instead of string formatting (default: False)
template_dir (str): Directory containing templates (default: 'templates/')
use_sudo (bool): Use sudo for upload (default: False)
backup (bool): Create backup of existing file (default: True)
mirror_local_mode (bool): Mirror local file permissions (default: False)
mode (str|int): Set specific file permissions
pty (bool): Request pseudo-terminal for command execution
Returns:
_AttributeList: Upload results
"""Usage Examples:
from fabric.contrib.files import upload_template
from fabric.api import env
# Simple template upload with string formatting
context = {
'server_name': 'myapp.example.com',
'document_root': '/var/www/myapp',
'port': 80
}
upload_template('nginx.conf.template', '/etc/nginx/sites-available/myapp',
context=context, use_sudo=True, backup=True)
# Jinja2 template with complex logic
jinja_context = {
'app_name': 'myapp',
'servers': ['web1.example.com', 'web2.example.com'],
'ssl_enabled': True,
'environment': env.environment
}
upload_template('app.conf.j2', '/etc/supervisor/conf.d/myapp.conf',
context=jinja_context, use_jinja=True, use_sudo=True)
# Template from custom directory
upload_template('database.yml', '/opt/myapp/config/database.yml',
context={'db_host': env.db_host, 'db_pass': env.db_password},
template_dir='deployment/templates/', mode='600')
# Example template content (nginx.conf.template):
# server {
# listen %(port)s;
# server_name %(server_name)s;
# root %(document_root)s;
#
# location / {
# try_files $uri $uri/ =404;
# }
# }In-place text file editing functions for search and replace, commenting, and content manipulation.
def sed(filename, before, after, limit='', use_sudo=False, backup='.bak', flags='', shell=False):
"""
Search and replace text in files using sed.
Args:
filename (str): File to modify
before (str): Pattern to search for (regex)
after (str): Replacement text
limit (str): Limit to specific line numbers
use_sudo (bool): Use sudo for file access (default: False)
backup (str): Backup file extension (default: '.bak')
flags (str): Additional sed flags
shell (bool): Use shell for command execution (default: False)
"""
def comment(filename, regex, use_sudo=False, char='#', backup='.bak', shell=False):
"""
Comment lines matching regex pattern.
Args:
filename (str): File to modify
regex (str): Pattern to match for commenting
use_sudo (bool): Use sudo for file access (default: False)
char (str): Comment character (default: '#')
backup (str): Backup file extension (default: '.bak')
shell (bool): Use shell for command execution (default: False)
"""
def uncomment(filename, regex, use_sudo=False, char='#', backup='.bak', shell=False):
"""
Uncomment lines matching regex pattern.
Args:
filename (str): File to modify
regex (str): Pattern to match for uncommenting
use_sudo (bool): Use sudo for file access (default: False)
char (str): Comment character to remove (default: '#')
backup (str): Backup file extension (default: '.bak')
shell (bool): Use shell for command execution (default: False)
"""Usage Examples:
from fabric.contrib.files import sed, comment, uncomment
from fabric.api import sudo
# Replace configuration values
sed('/etc/nginx/nginx.conf',
'worker_processes auto;',
'worker_processes 4;',
use_sudo=True)
# Update database connection string
sed('/opt/myapp/config.py',
r'DATABASE_URL = .*',
r'DATABASE_URL = "postgresql://user:pass@localhost/myapp"',
use_sudo=True, backup='.backup')
# Comment out debug settings
comment('/etc/ssh/sshd_config',
r'^PasswordAuthentication yes',
use_sudo=True)
# Uncomment production settings
uncomment('/etc/php/7.4/apache2/php.ini',
r'^;opcache.enable=1',
char=';', use_sudo=True)
# Multiple replacements with sed flags
sed('/var/log/app.log',
r'ERROR',
'WARNING',
flags='g', # Global replacement
backup='') # No backupFunctions for checking file contents and appending text to files.
def contains(filename, text, exact=False, use_sudo=False, escape=True, shell=False):
"""
Check if file contains specific text.
Args:
filename (str): File to check
text (str): Text to search for
exact (bool): Exact string match vs regex (default: False)
use_sudo (bool): Use sudo for file access (default: False)
escape (bool): Escape regex special characters (default: True)
shell (bool): Use shell for command execution (default: False)
Returns:
bool: True if text found in file
"""
def append(filename, text, use_sudo=False, partial=False, escape=True, shell=False):
"""
Append text to file if not already present.
Args:
filename (str): File to modify
text (str): Text to append
use_sudo (bool): Use sudo for file access (default: False)
partial (bool): Allow partial line matches (default: False)
escape (bool): Escape regex special characters (default: True)
shell (bool): Use shell for command execution (default: False)
"""Usage Examples:
from fabric.contrib.files import contains, append
from fabric.api import sudo
# Check if configuration exists
if contains('/etc/hosts', 'myapp.local'):
print("Host entry already exists")
else:
append('/etc/hosts', '127.0.0.1 myapp.local', use_sudo=True)
# Add SSH key if not present
ssh_key = 'ssh-rsa AAAAB3... user@example.com'
if not contains('/home/deploy/.ssh/authorized_keys', ssh_key):
append('/home/deploy/.ssh/authorized_keys', ssh_key, use_sudo=True)
# Add cron job if not already scheduled
cron_line = '0 2 * * * /opt/myapp/backup.sh'
if not contains('/var/spool/cron/crontabs/root', cron_line, use_sudo=True):
append('/var/spool/cron/crontabs/root', cron_line, use_sudo=True)
# Check for exact string match
if contains('/proc/version', 'Ubuntu', exact=True):
print("Running on Ubuntu")
# Append with partial matching (substring check)
append('/etc/environment',
'PATH="/usr/local/bin:$PATH"',
partial=True, use_sudo=True)High-level functions for synchronizing entire projects and directories using rsync and tar-based transfers.
def rsync_project(remote_dir, local_dir=None, exclude=(), delete=False, extra_opts='', ssh_opts='', capture=False, upload=True, default_opts='-pthrvz'):
"""
Synchronize project files via rsync.
Args:
remote_dir (str): Remote directory path
local_dir (str): Local directory path (default: current directory)
exclude (tuple): Patterns to exclude from sync
delete (bool): Delete remote files not in local (default: False)
extra_opts (str): Additional rsync options
ssh_opts (str): SSH connection options
capture (bool): Capture output instead of printing (default: False)
upload (bool): Upload to remote (True) or download (False)
default_opts (str): Default rsync options (default: '-pthrvz')
Returns:
_AttributeString: rsync command output
"""
def upload_project(local_dir=None, remote_dir="", use_sudo=False):
"""
Upload project directory via tar/gzip compression.
Args:
local_dir (str): Local project directory (default: current directory)
remote_dir (str): Remote destination directory
use_sudo (bool): Use sudo for remote operations (default: False)
Returns:
_AttributeString: Upload operation result
"""Usage Examples:
from fabric.contrib.project import rsync_project, upload_project
from fabric.api import cd, run
# Basic project sync
rsync_project('/var/www/myapp/', 'dist/')
# Sync with exclusions and delete
rsync_project('/var/www/myapp/',
exclude=('node_modules/', '.git/', '*.log'),
delete=True)
# Download from remote (backup)
rsync_project('/var/www/myapp/', 'backup/', upload=False)
# Custom rsync options
rsync_project('/opt/myapp/',
extra_opts='--compress-level=9 --progress',
ssh_opts='-o StrictHostKeyChecking=no')
# Upload entire project as tar.gz
upload_project('myapp/', '/tmp/deployments/', use_sudo=True)
# Typical deployment workflow
def deploy():
# Build locally
local('npm run build')
# Sync to remote
rsync_project('/var/www/myapp/', 'dist/',
exclude=('*.map', 'tests/'),
delete=True)
# Restart services
with cd('/var/www/myapp'):
sudo('systemctl reload nginx')
# Large project upload with compression
def deploy_large_project():
# Create compressed archive locally
local('tar czf myapp.tar.gz --exclude=node_modules .')
# Upload and extract
put('myapp.tar.gz', '/tmp/', use_sudo=True)
with cd('/var/www'):
sudo('tar xzf /tmp/myapp.tar.gz')
sudo('rm /tmp/myapp.tar.gz')
local('rm myapp.tar.gz') # Cleanup local archiveBoth put() and get() return _AttributeList objects that provide detailed information about transfer operations:
from fabric.api import put, get
# Upload multiple files
results = put('config/*', '/etc/myapp/', use_glob=True)
# Check overall success
if results.succeeded:
print(f"Successfully uploaded {len(results)} files")
# Handle partial failures
if results.failed:
print("Some uploads failed:")
for failed_file in results.failed:
print(f" Failed: {failed_file}")
# Process successful uploads
for uploaded_file in results:
if uploaded_file not in results.failed:
print(f" Success: {uploaded_file}")
# Download with error handling
downloads = get('/var/log/*.log', use_glob=True)
if not downloads.succeeded:
print("Download failed completely")
elif downloads.failed:
print(f"Partial failure: {len(downloads.failed)} files failed")Interactive functions for user confirmation during file operations and deployment workflows.
def confirm(question, default=True):
"""
Ask user a yes/no question and return True or False.
Args:
question (str): Question to ask the user
default (bool): Default value if user presses enter (default: True)
Returns:
bool: True if user confirms, False otherwise
"""Usage Examples:
from fabric.contrib.console import confirm
from fabric.api import run, sudo
# Confirm dangerous operations
if confirm('Delete all log files?', default=False):
sudo('rm -rf /var/log/*.log')
# Confirm deployment
if confirm('Deploy to production?'):
run('git pull origin master')
sudo('systemctl restart nginx')
else:
print("Deployment cancelled")
# Use in deployment scripts
@task
def deploy():
if not confirm('Deploy to production? This cannot be undone.', default=False):
abort('Deployment cancelled by user')
# Continue with deployment
rsync_project('/var/www/myapp/', 'dist/')
sudo('systemctl reload nginx')File operations work seamlessly with Fabric's context managers:
from fabric.api import *
from fabric.contrib.files import *
# Upload within directory context
with cd('/var/www'):
put('app.tar.gz', '.')
run('tar xzf app.tar.gz')
# Use sudo context for protected operations
with settings(use_sudo=True):
upload_template('nginx.conf', '/etc/nginx/sites-available/myapp')
append('/etc/hosts', '127.0.0.1 myapp.local')
# Combine multiple contexts
with cd('/opt/myapp'), settings(use_sudo=True):
if not exists('config.json'):
upload_template('config.json.template', 'config.json',
context={'debug': False})Install with Tessl CLI
npx tessl i tessl/pypi-fabric