Error Handling

class ErrorReturnCode(Exception):
    """
    Base exception for all command execution errors.
    
    Attributes:
    - full_cmd: str = complete command that was executed
    - stdout: bytes = command's stdout output
    - stderr: bytes = command's stderr output  
    - truncate_cap: int = output truncation limit
    """
    
    def __init__(self, full_cmd: str, stdout: bytes, stderr: bytes, truncate_cap: int = 750): ...
    
    @property
    def exit_code(self) -> int:
        """The exit code returned by the command."""

class ErrorReturnCode_1(ErrorReturnCode):
    """Exception for commands that exit with code 1 (general errors)."""

class ErrorReturnCode_2(ErrorReturnCode):  
    """Exception for commands that exit with code 2 (misuse of shell builtins)."""

class ErrorReturnCode_126(ErrorReturnCode):
    """Exception for commands that exit with code 126 (command not executable)."""

class ErrorReturnCode_127(ErrorReturnCode):
    """Exception for commands that exit with code 127 (command not found)."""

class ErrorReturnCode_128(ErrorReturnCode):
    """Exception for commands that exit with code 128 (invalid exit argument)."""

# Note: sh dynamically creates ErrorReturnCode_N classes for exit codes 1-255
# Common exit codes include:
# 1 - General errors, 2 - Misuse of shell builtins, 126 - Command not executable
# 127 - Command not found, 128 - Invalid exit argument, 130 - Script terminated by Ctrl+C

class SignalException(Exception):
    """Exception raised when process is terminated by a signal."""
    
    def __init__(self, full_cmd: str, signal_code: int): ...

class TimeoutException(Exception):
    """Exception raised when command times out."""
    
    def __init__(self, full_cmd: str, timeout: float): ...

class CommandNotFound(Exception):
    """Exception raised when command cannot be found in PATH."""
    
    def __init__(self, command: str): ...

class ForkException(Exception):
    """Exception raised when there's an error in the fork process."""
    
    def __init__(self, command: str, error: str): ...

import sh

# Catch specific exit codes
try:
    sh.grep("nonexistent_pattern", "file.txt")
except sh.ErrorReturnCode_1:
    print("Pattern not found (exit code 1)")
except sh.ErrorReturnCode_2:
    print("File not found or other error (exit code 2)")

# Catch any command error
try:
    result = sh.ls("/nonexistent_directory")
except sh.ErrorReturnCode as e:
    print(f"Command failed: {e.full_cmd}")
    print(f"Exit code: {e.exit_code}")
    print(f"Stderr: {e.stderr.decode('utf-8')}")

# Handle signal termination
try:
    sh.sleep(60, _timeout=5)
except sh.TimeoutException as e:
    print(f"Command timed out after {e.timeout} seconds")
except sh.SignalException as e:
    print(f"Command terminated by signal {e.signal_code}")

def __call__(self, *args, _ok_code=None, **kwargs):
    """
    Execute command with custom success codes.
    
    Parameters:
    - _ok_code: int/list = exit codes to treat as success
    
    Returns:
    str: Command output
    """

import sh

# Accept multiple exit codes as success
try:
    # grep returns 0 if found, 1 if not found, 2 for errors
    result = sh.grep("pattern", "file.txt", _ok_code=[0, 1])
    if "pattern" in result:
        print("Pattern found")
    else:
        print("Pattern not found (but that's OK)")
except sh.ErrorReturnCode:
    print("Real error occurred (exit code 2)")

# Accept any exit code
output = sh.some_command(_ok_code=list(range(256)))

# Custom validation logic
def validate_exit_code(code, stdout, stderr):
    """Custom logic to determine if command succeeded."""
    if code == 0:
        return True
    if code == 1 and "warning" in stderr.decode().lower():
        return True  # Treat warnings as success
    return False

# Use with manual error handling
proc = sh.custom_tool(_bg=True)
proc.wait()

if validate_exit_code(proc.exit_code, proc.stdout, proc.stderr):
    print("Command succeeded (with custom logic)")
else:
    print(f"Command failed with exit code {proc.exit_code}")

class ErrorReturnCode:
    @property
    def exit_code(self) -> int:
        """Exit code returned by the command."""
    
    @property  
    def full_cmd(self) -> str:
        """Complete command line that was executed."""
    
    @property
    def stdout(self) -> bytes:
        """Standard output from the failed command."""
    
    @property
    def stderr(self) -> bytes:
        """Standard error from the failed command."""

import sh
import logging

# Detailed error logging
logger = logging.getLogger(__name__)

try:
    sh.complex_command("arg1", "arg2", "--option", "value")
except sh.ErrorReturnCode as e:
    logger.error(f"Command failed: {e.full_cmd}")
    logger.error(f"Exit code: {e.exit_code}")
    
    # Log stdout if available
    if e.stdout:
        logger.info(f"Stdout: {e.stdout.decode('utf-8', errors='replace')}")
    
    # Log stderr if available  
    if e.stderr:
        logger.error(f"Stderr: {e.stderr.decode('utf-8', errors='replace')}")
    
    # Re-raise for higher-level handling
    raise

# Error analysis
def analyze_error(error):
    """Analyze error and suggest solutions."""
    if isinstance(error, sh.CommandNotFound):
        return f"Command '{error.command}' not found. Check if it's installed and in PATH."
    
    if isinstance(error, sh.TimeoutException):
        return f"Command timed out after {error.timeout} seconds. Consider increasing timeout."
    
    if isinstance(error, sh.ErrorReturnCode):
        stderr_text = error.stderr.decode('utf-8', errors='replace').lower()
        
        if "permission denied" in stderr_text:
            return "Permission denied. Try running with sudo or check file permissions."
        elif "no such file" in stderr_text:
            return "File or directory not found. Check the path."
        elif "command not found" in stderr_text:
            return "Command not found in PATH."
        else:
            return f"Command failed with exit code {error.exit_code}"
    
    return "Unknown error occurred."

try:
    sh.restricted_command()
except Exception as e:
    suggestion = analyze_error(e)
    print(f"Error: {suggestion}")

import sh
import time
import random

def retry_command(command_func, max_retries=3, backoff_factor=1.0):
    """
    Retry a command with exponential backoff.
    
    Parameters:
    - command_func: callable that executes the sh command
    - max_retries: int = maximum number of retry attempts
    - backoff_factor: float = multiplier for retry delay
    
    Returns:
    str: Command output if successful
    
    Raises:
    Exception: Last exception if all retries fail
    """
    last_exception = None
    
    for attempt in range(max_retries + 1):
        try:
            return command_func()
        except (sh.ErrorReturnCode, sh.TimeoutException) as e:
            last_exception = e
            
            if attempt < max_retries:
                delay = (backoff_factor * (2 ** attempt)) + random.uniform(0, 1)
                print(f"Attempt {attempt + 1} failed, retrying in {delay:.2f}s...")
                time.sleep(delay)
            else:
                print(f"All {max_retries + 1} attempts failed")
    
    raise last_exception

# Usage examples
def flaky_network_command():
    return sh.curl("http://unreliable-server.com/api", _timeout=10)

try:
    result = retry_command(flaky_network_command, max_retries=3, backoff_factor=2.0)
    print("Command succeeded:", result[:100])
except Exception as e:
    print(f"Command failed after retries: {e}")

# Selective retry based on error type
def smart_retry_command(command_func, max_retries=3):
    """Retry only for certain types of errors."""
    retryable_errors = (sh.TimeoutException, sh.ErrorReturnCode_124)  # timeout error codes
    
    for attempt in range(max_retries + 1):
        try:
            return command_func()
        except retryable_errors as e:
            if attempt < max_retries:
                print(f"Retryable error on attempt {attempt + 1}: {e}")
                time.sleep(2 ** attempt)  # Exponential backoff
            else:
                raise
        except Exception as e:
            # Don't retry for non-retryable errors
            print(f"Non-retryable error: {e}")
            raise

# Fallback command strategy
def command_with_fallback(primary_cmd, fallback_cmd):
    """Try primary command, fall back to alternative if it fails."""
    try:
        return primary_cmd()
    except sh.CommandNotFound:
        print("Primary command not found, trying fallback...")
        return fallback_cmd()
    except sh.ErrorReturnCode as e:
        if e.exit_code in [126, 127]:  # Permission or not found
            print("Primary command failed, trying fallback...")
            return fallback_cmd()
        else:
            raise

# Usage
result = command_with_fallback(
    lambda: sh.gls("-la"),  # GNU ls
    lambda: sh.ls("-la")   # BSD ls fallback
)

import sh
from contextlib import contextmanager

@contextmanager
def ignore_errors(*error_types):
    """Context manager to ignore specific error types."""
    try:
        yield
    except error_types as e:
        print(f"Ignoring error: {e}")

@contextmanager  
def log_errors(logger):
    """Context manager to log but not raise errors."""
    try:
        yield
    except Exception as e:
        logger.error(f"Command error: {e}")
        # Don't re-raise, just log

# Usage examples
import logging
logger = logging.getLogger(__name__)

# Ignore file not found errors
with ignore_errors(sh.ErrorReturnCode_2):
    sh.rm("nonexistent_file.txt")
    print("File deletion attempted (may not have existed)")

# Log errors but continue
with log_errors(logger):
    sh.risky_command()
    print("Risky command attempted")

# Custom error context
class CommandContext:
    def __init__(self, description):
        self.description = description
    
    def __enter__(self):
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type and issubclass(exc_type, sh.ErrorReturnCode):
            print(f"Error in {self.description}: {exc_val}")
            return True  # Suppress the exception
        return False

# Usage
with CommandContext("database backup"):
    sh.mysqldump("--all-databases")
    
with CommandContext("log rotation"):
    sh.logrotate("/etc/logrotate.conf")