Utilities

@contextmanager
def pushd(path):
    """
    Context manager to temporarily change working directory.
    
    Parameters:
    - path: str/Path = directory to change to
    
    Yields:
    str: The new current working directory
    
    Raises:
    OSError: If directory doesn't exist or can't be accessed
    """

import sh

# Temporary directory change
original_dir = sh.pwd().strip()
print(f"Starting in: {original_dir}")

with sh.pushd("/tmp"):
    current = sh.pwd().strip()
    print(f"Now in: {current}")
    
    # All commands run in /tmp
    files = sh.ls("-la")
    sh.touch("test_file.txt")

# Automatically back to original directory
back_dir = sh.pwd().strip()
print(f"Back in: {back_dir}")
assert back_dir == original_dir

# Nested directory changes
with sh.pushd("/tmp"):
    print(f"Level 1: {sh.pwd().strip()}")
    
    with sh.pushd("subdir"):
        print(f"Level 2: {sh.pwd().strip()}")
        sh.touch("nested_file.txt")
    
    print(f"Back to level 1: {sh.pwd().strip()}")

print(f"Back to original: {sh.pwd().strip()}")

# Error handling with pushd
try:
    with sh.pushd("/nonexistent"):
        print("This won't execute")
except OSError as e:
    print(f"Directory change failed: {e}")

def glob(pattern, *args, **kwargs):
    """
    Enhanced glob function that returns results compatible with sh commands.
    
    Parameters:
    - pattern: str = glob pattern (supports *, ?, [], {})
    - *args: additional patterns to match
    - **kwargs: glob options
    
    Returns:
    GlobResults: List-like object with enhanced functionality
    """

class GlobResults(list):
    """Enhanced list of glob results with additional methods."""
    def __init__(self, results): ...

import sh

# Basic globbing
log_files = sh.glob("/var/log/*.log")
print(f"Found {len(log_files)} log files")

# Use glob results with sh commands
for log_file in log_files:
    size = sh.wc("-l", log_file)
    print(f"{log_file}: {size.strip()} lines")

# Multiple patterns
config_files = sh.glob("*.conf", "*.cfg", "*.ini")
print("Configuration files:", config_files)

# Complex patterns
python_files = sh.glob("**/*.py", recursive=True)
test_files = sh.glob("**/test_*.py", "**/tests/*.py")

# Integration with other commands
sh.tar("czf", "backup.tar.gz", *sh.glob("*.txt", "*.md"))

# Filter glob results
large_files = []
for file_path in sh.glob("*"):
    try:
        size = int(sh.stat("-c", "%s", file_path))
        if size > 1024 * 1024:  # > 1MB
            large_files.append(file_path)
    except sh.ErrorReturnCode:
        pass  # Skip files we can't stat

print(f"Large files: {large_files}")

class Logger:
    """
    Memory-efficient command execution logger.
    
    Attributes:
    - name: str = logger name
    - context: dict = additional context information
    """
    
    def __init__(self, name: str, context: dict = None): ...
    
    def log(self, level: str, message: str): ...
    def info(self, message: str): ...
    def debug(self, message: str): ...
    def warning(self, message: str): ...
    def error(self, message: str): ...

def __call__(self, *args, _log=None, **kwargs):
    """
    Execute command with logging.
    
    Parameters:
    - _log: Logger/bool = logger instance or True for default logging
    
    Returns:
    str: Command output
    """

import sh
import logging

# Enable default logging
logging.basicConfig(level=logging.DEBUG)

# Log all command executions
result = sh.ls("-la", _log=True)

# Custom logger
custom_logger = sh.Logger("deployment", {"version": "1.2.3", "env": "production"})

sh.git("pull", _log=custom_logger)
sh.docker("build", "-t", "myapp:latest", ".", _log=custom_logger)
sh.docker("push", "myapp:latest", _log=custom_logger)

# Conditional logging
debug_mode = True
if debug_mode:
    logger = sh.Logger("debug")
else:
    logger = None

sh.make("build", _log=logger)

# Log analysis
class CommandAuditor:
    def __init__(self):
        self.commands = []
        self.logger = sh.Logger("auditor")
        self.logger.log = self._capture_log
    
    def _capture_log(self, level, message):
        self.commands.append({
            'level': level,
            'message': message,
            'timestamp': time.time()
        })
    
    def get_summary(self):
        return {
            'total_commands': len(self.commands),
            'errors': len([c for c in self.commands if c['level'] == 'error']),
            'commands': self.commands
        }

auditor = CommandAuditor()
sh.ls(_log=auditor.logger)
sh.pwd(_log=auditor.logger)
print(auditor.get_summary())

class Command:
    def __str__(self) -> str:
        """String representation showing executable path and baked arguments."""
    
    def __repr__(self) -> str:
        """Formal string representation of the Command object."""
    
    def __eq__(self, other) -> bool:
        """Compare Command objects for equality."""

import sh

# Command discovery and validation
def validate_dependencies(commands):
    """Check if required commands are available."""
    missing = []
    available = []
    
    for cmd_name in commands:
        try:
            cmd = sh.Command(cmd_name)
            available.append({
                'name': cmd_name,
                'command': str(cmd),
                'version': get_version(cmd)
            })
        except sh.CommandNotFound:
            missing.append(cmd_name)
    
    return available, missing

def get_version(cmd):
    """Try to get version information for a command."""
    version_flags = ['--version', '-V', '-v']
    
    for flag in version_flags:
        try:
            output = cmd(flag)
            return output.split('\n')[0]  # First line usually has version
        except sh.ErrorReturnCode:
            continue
    
    return "unknown"

# Check deployment dependencies
required_tools = ['git', 'docker', 'kubectl', 'helm']
available, missing = validate_dependencies(required_tools)

print("Available tools:")
for tool in available:
    print(f"  {tool['name']}: {tool['command']} ({tool['version']})")

if missing:
    print(f"Missing tools: {missing}")
    exit(1)

# Command introspection
git_cmd = sh.git
print(f"Git command: {git_cmd}")
print(f"Git string representation: {str(git_cmd)}")
print(f"Git repr: {repr(git_cmd)}")

# Command comparison
def compare_commands(*cmd_names):
    """Compare different commands and their capabilities."""
    results = {}
    
    for name in cmd_names:
        try:
            cmd = sh.Command(name)
            cmd_str = str(cmd)
            results[name] = {
                'available': True,
                'command': cmd_str,
                'repr': repr(cmd)
            }
        except sh.CommandNotFound:
            results[name] = {'available': False}
    
    return results

# Compare different versions of python
python_variants = compare_commands('python', 'python3', 'python3.9', 'python3.10')
for name, info in python_variants.items():
    if info['available']:
        print(f"{name}: {info['command']}")
    else:
        print(f"{name}: not available")

class StreamBufferer:
    """
    Internal buffering implementation for command I/O streams.
    
    Most users don't need this - it's primarily used internally by sh
    but available for advanced buffering control scenarios.
    
    Parameters:
    - buffer_size: int = 0 (unbuffered), 1 (line-buffered), N (N-byte buffered)
    - encoding: str = character encoding for text processing  
    """
    
    def __init__(self, buffer_size: int = 0, encoding: str = 'utf-8'): ...

import sh

# Most users should use standard command execution
# This is only for advanced scenarios requiring custom buffering control
bufferer = sh.StreamBufferer(buffer_size=1024, encoding='utf-8')

# Note: StreamBufferer is primarily an internal implementation detail
# Standard sh command execution handles buffering automatically

def __call__(self, *args, _env=None, _cwd=None, **kwargs):
    """
    Execute command with environment customization.
    
    Parameters:
    - _env: dict = environment variables to set
    - _cwd: str = working directory for command
    
    Returns:
    str: Command output
    """

import sh
import os

# Environment management
def with_env(**env_vars):
    """Context manager for temporary environment variables."""
    old_env = {}
    
    try:
        # Set new environment variables
        for key, value in env_vars.items():
            old_env[key] = os.environ.get(key)
            os.environ[key] = str(value)
        yield
    finally:
        # Restore old environment
        for key, old_value in old_env.items():
            if old_value is None:
                os.environ.pop(key, None)
            else:
                os.environ[key] = old_value

# Usage
with with_env(DEBUG=1, LOG_LEVEL="verbose"):
    sh.my_app("--config", "production.conf")

# Per-command environment
current_env = os.environ.copy()
current_env.update({
    'PYTHONPATH': '/custom/path',
    'DEBUG': '1'
})

result = sh.python("script.py", _env=current_env)

# Configuration helpers
class ConfigManager:
    def __init__(self, config_file="config.json"):
        self.config = self.load_config(config_file)
    
    def load_config(self, file_path):
        """Load configuration from file."""
        try:
            import json
            with open(file_path) as f:
                return json.load(f)
        except FileNotFoundError:
            return {}
    
    def get_env_for_service(self, service_name):
        """Get environment variables for a specific service."""
        service_config = self.config.get(service_name, {})
        env = os.environ.copy()
        env.update(service_config.get('env', {}))
        return env
    
    def run_service_command(self, service_name, command, *args):
        """Run a command with service-specific configuration."""
        env = self.get_env_for_service(service_name)
        cwd = self.config.get(service_name, {}).get('working_dir')
        
        return command(*args, _env=env, _cwd=cwd)

# Usage
config = ConfigManager("services.json")
result = config.run_service_command("web", sh.npm, "start")

# Path management utilities
def ensure_command_available(cmd_name, install_hint=None):
    """Ensure a command is available, provide installation hint if not."""
    try:
        cmd = sh.Command(cmd_name)
        return cmd
    except sh.CommandNotFound:
        hint = install_hint or f"Install {cmd_name} to continue"
        raise sh.CommandNotFound(f"{cmd_name} not found. {hint}")

# Usage
git = ensure_command_available('git', 'Install git: sudo apt-get install git')
docker = ensure_command_available('docker', 'Install Docker: https://docs.docker.com/install/')

# Environment variables are accessible as module attributes
# sh.ENVIRONMENT_VARIABLE returns the value of $ENVIRONMENT_VARIABLE

import sh
import os

# Access environment variables via sh module
home_dir = sh.HOME
print(f"Home directory: {home_dir}")

# Access PATH environment variable
path_var = sh.PATH
print(f"PATH: {path_var}")

# Check if environment variable exists
try:
    custom_var = sh.MY_CUSTOM_VAR
    print(f"Custom variable: {custom_var}")
except sh.CommandNotFound:
    print("MY_CUSTOM_VAR not set")

# Compare with os.environ access
print("Via sh.HOME:", sh.HOME)
print("Via os.environ:", os.environ.get('HOME'))

# Dynamic environment variable access
def get_env_var(var_name):
    """Get environment variable via sh module."""
    try:
        return getattr(sh, var_name)
    except sh.CommandNotFound:
        return None

user = get_env_var('USER')
shell = get_env_var('SHELL') 
editor = get_env_var('EDITOR')

print(f"User: {user}, Shell: {shell}, Editor: {editor}")

def _args(**kwargs):
    """
    Deprecated context manager for setting default command arguments.
    
    Note: This is deprecated. Consider using sh.bake() or Command.bake() instead.
    
    Parameters:
    - **kwargs: Default arguments to apply to all commands in context
    
    Returns:
    ContextManager: Context manager for temporary default arguments
    """

def bake(**kwargs):
    """
    Create a new sh module instance with baked-in default arguments.
    
    Parameters:
    - **kwargs: Default execution options for all commands
    
    Returns:
    SelfWrapper: New sh module instance with default arguments
    """

import sh

# Module-level baking (recommended approach)
sh_verbose = sh.bake(_out=lambda line: print(f"CMD: {line.strip()}"))

# All commands through sh_verbose will have verbose output
sh_verbose.ls("-la")
sh_verbose.pwd()

# Create sh instance with specific environment
production_sh = sh.bake(_env={"ENVIRONMENT": "production"})
production_sh.deploy_script()

# Create sh instance with common options
background_sh = sh.bake(_bg=True)
proc1 = background_sh.long_running_task()
proc2 = background_sh.another_task()

# Legacy _args usage (deprecated, avoid in new code)
with sh._args(_timeout=30):
    # All commands in this block have 30-second timeout
    sh.curl("http://slow-server.com")
    sh.wget("http://large-file.com/download")

# Preferred modern approach using bake
timeout_sh = sh.bake(_timeout=30)
timeout_sh.curl("http://slow-server.com")
timeout_sh.wget("http://large-file.com/download")