tessl/pypi-parallel-ssh

Asynchronous parallel SSH client library that enables developers to execute SSH commands across many servers simultaneously with minimal system load on the client host.

—

Pending

Overview

Eval results

Files

Single Host SSH Operations

Name: tessl/pypi-parallel-ssh
Author: tessl

Connect to and execute commands on individual hosts with the same rich feature set as parallel operations. Ideal for single-host workflows, debugging, or when parallel execution is not needed.

Capabilities

SSHClient Initialization

Create an SSH client for connecting to a single host with comprehensive authentication and connection options.

class SSHClient:
    def __init__(self, host, user=None, password=None, port=None, pkey=None,
                 alias=None, num_retries=DEFAULT_RETRIES, retry_delay=RETRY_DELAY,
                 allow_agent=True, timeout=None, forward_ssh_agent=False,
                 proxy_host=None, proxy_port=None, proxy_pkey=None,
                 proxy_user=None, proxy_password=None, _auth_thread_pool=True,
                 keepalive_seconds=60, identity_auth=True, ipv6_only=False):
        """
        Create SSH client for single host.

        Parameters:
        - host (str): Hostname or IP address
        - user (str, optional): Username for authentication
        - password (str, optional): Password for authentication  
        - port (int, optional): SSH port number (default: 22)
        - pkey (str or bytes, optional): Private key file path or key data
        - alias (str, optional): Host alias for identification
        - num_retries (int, optional): Connection retry attempts (default: 3)
        - retry_delay (float, optional): Delay between retries (default: 5)
        - allow_agent (bool, optional): Use SSH agent authentication (default: True)
        - timeout (float, optional): Session timeout in seconds
        - forward_ssh_agent (bool, optional): SSH agent forwarding (default: False)
        - proxy_host (str, optional): SSH proxy hostname
        - proxy_port (int, optional): SSH proxy port
        - proxy_pkey (str or bytes, optional): SSH proxy private key
        - proxy_user (str, optional): SSH proxy username
        - proxy_password (str, optional): SSH proxy password
        - _auth_thread_pool (bool, optional): Use thread pool for authentication (default: True)
        - keepalive_seconds (int, optional): Keepalive interval (default: 60)
        - identity_auth (bool, optional): Identity file authentication (default: True)
        - ipv6_only (bool, optional): IPv6 only mode (default: False)
        """

Usage examples:

from pssh.clients import SSHClient

# Simple connection
client = SSHClient('server.example.com', user='admin')

# With private key authentication
client = SSHClient('server.example.com', user='admin', pkey='~/.ssh/id_rsa')

# Through SSH proxy
client = SSHClient(
    'internal-server.example.com',
    user='admin',
    proxy_host='bastion.example.com',
    proxy_user='proxy_user'
)

# With custom port and alias
client = SSHClient(
    '192.168.1.100',
    port=2222,
    user='admin',
    alias='dev-server'
)

Single Host Command Execution

Execute commands on the connected host with options for sudo, user switching, and output handling.

def run_command(self, command, sudo=False, user=None, use_pty=False,
               shell=None, encoding='utf-8', timeout=None, read_timeout=None):
    """
    Execute command on the host.

    Parameters:
    - command (str): Command to execute
    - sudo (bool, optional): Run command with sudo (default: False)
    - user (str, optional): Run command as different user
    - use_pty (bool, optional): Use pseudo-terminal (default: False)
    - shell (str, optional): Shell to use for command execution
    - encoding (str, optional): Output encoding (default: 'utf-8')
    - timeout (float, optional): Command execution timeout
    - read_timeout (float, optional): Output read timeout

    Returns:
    HostOutput: Object containing command results and output streams
    """

Usage examples:

# Basic command execution
host_output = client.run_command('ls -la /home')

# Process output
for line in host_output.stdout:
    print(line)
print(f"Exit code: {host_output.exit_code}")

# Run with sudo
host_output = client.run_command('systemctl restart nginx', sudo=True)

# Run as different user
host_output = client.run_command('whoami', user='www-data')

# Long running command with timeout
try:
    host_output = client.run_command('sleep 30', timeout=10)
except Timeout:
    print("Command timed out")

Session Management

Manage SSH sessions and channels for more granular control over connections.

def open_session(self):
    """
    Open a new SSH channel session.

    Returns:
    ssh2.channel.Channel: New SSH channel for command execution
    """

def close_channel(self, channel):
    """
    Close an SSH channel.

    Parameters:
    - channel: SSH channel to close
    """

def get_exit_status(self, channel):
    """
    Get exit status from a channel.

    Parameters:
    - channel: SSH channel to check

    Returns:
    int or None: Exit status if available, None if not ready
    """

def finished(self, channel):
    """
    Check if command on channel has finished.

    Parameters:
    - channel: SSH channel to check

    Returns:
    bool: True if command finished, False otherwise
    """

Usage example:

# Manual session management
channel = client.open_session()
channel.execute('long_running_command')

# Check periodically if finished
import time
while not client.finished(channel):
    time.sleep(1)
    print("Still running...")

exit_code = client.get_exit_status(channel)
client.close_channel(channel)
print(f"Command finished with exit code: {exit_code}")

Result Synchronization

Wait for command completion with timeout handling.

def wait_finished(self, host_output, timeout=None):
    """
    Wait for command in host_output to complete.

    Parameters:
    - host_output (HostOutput): Host output object to wait for
    - timeout (float, optional): Timeout in seconds

    Raises:
    - Timeout: If timeout is reached before completion
    """

Usage example:

# Execute and wait for completion
host_output = client.run_command('backup_database.sh')

try:
    client.wait_finished(host_output, timeout=300)  # 5 minute timeout
    print("Backup completed successfully")
    print(f"Exit code: {host_output.exit_code}")
except Timeout:
    print("Backup did not complete within 5 minutes")

Advanced Buffer and Output Management

Manage low-level output buffers and read operations for fine-grained control over command output processing.

def read_output(self, stdout_buffer, timeout=None):
    """
    Read output from stdout buffer.

    Parameters:
    - stdout_buffer: Output buffer to read from
    - timeout (float, optional): Read timeout in seconds

    Returns:
    generator: Lines from stdout buffer
    """

def read_stderr(self, stderr_buffer, timeout=None):
    """
    Read output from stderr buffer.

    Parameters:
    - stderr_buffer: Error buffer to read from  
    - timeout (float, optional): Read timeout in seconds

    Returns:
    generator: Lines from stderr buffer
    """

def read_output_buffer(self, output_buffer, prefix=None, callback=None,
                      callback_args=None, encoding='utf-8'):
    """
    Read from output buffer with optional processing.

    Parameters:
    - output_buffer: Buffer to read from
    - prefix (str, optional): Prefix for output lines
    - callback (callable, optional): Callback function for each line
    - callback_args (tuple, optional): Arguments for callback function
    - encoding (str, optional): Text encoding (default: 'utf-8')

    Returns:
    generator: Processed output lines
    """

SFTP Low-Level Operations

Direct SFTP operations for advanced use cases requiring fine-grained control over file transfers.

def sftp_put(self, sftp, local_file, remote_file):
    """
    Upload file via SFTP session.

    Parameters:
    - sftp: Active SFTP session object
    - local_file (str): Local file path
    - remote_file (str): Remote destination path
    """

def sftp_get(self, sftp, remote_file, local_file):
    """
    Download file via SFTP session.

    Parameters:
    - sftp: Active SFTP session object
    - remote_file (str): Remote file path
    - local_file (str): Local destination path
    """

def mkdir(self, sftp, directory):
    """
    Create directory via SFTP.

    Parameters:
    - sftp: Active SFTP session object
    - directory (str): Directory path to create
    """

Error Handling

The single host client provides comprehensive error handling:

from pssh.exceptions import (
    UnknownHostError, ConnectionError, AuthenticationError,
    SessionError, Timeout, PKeyFileError
)

try:
    client = SSHClient('unreachable-host.example.com')
    host_output = client.run_command('echo test')
except UnknownHostError as e:
    print(f"Host not found: {e}")
except ConnectionError as e:
    print(f"Connection failed: {e}")
except AuthenticationError as e:
    print(f"Authentication failed: {e}")
except SessionError as e:
    print(f"Session error: {e}")
except PKeyFileError as e:
    print(f"Private key error: {e}")

Authentication Methods

The SSHClient supports multiple authentication methods:

# Password authentication
client = SSHClient('host.example.com', user='admin', password='secret')

# Private key file authentication
client = SSHClient('host.example.com', user='admin', pkey='/path/to/key')

# Private key data authentication  
with open('/path/to/key', 'rb') as f:
    key_data = f.read()
client = SSHClient('host.example.com', user='admin', pkey=key_data)

# SSH agent authentication (default)
client = SSHClient('host.example.com', user='admin', allow_agent=True)

# Disable SSH agent and identity files
client = SSHClient(
    'host.example.com', 
    user='admin',
    allow_agent=False,
    identity_auth=False,
    password='secret'
)

Connection Reuse

For multiple operations on the same host, reuse the client instance to avoid reconnection overhead:

client = SSHClient('host.example.com', user='admin')

# Multiple operations on same connection
host_output1 = client.run_command('uptime')
host_output2 = client.run_command('df -h') 
host_output3 = client.run_command('ps aux | head -10')

# Process all results
for i, output in enumerate([host_output1, host_output2, host_output3], 1):
    print(f"Command {i} output:")
    for line in output.stdout:
        print(f"  {line}")

Install with Tessl CLI