or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

docs

core-execution.mdindex.mdprivilege-elevation.mdprocess-management.mdutilities.md
tile.json

tessl/pypi-command-runner

Platform agnostic command and shell execution tool with timeout handling, live output capture, and UAC/sudo privilege elevation

Workspace
tessl
Visibility
Public
Created
Last updated
Describes
pypipkg:pypi/command-runner@1.7.x

To install, run

npx @tessl/cli install tessl/pypi-command-runner@1.7.0

index.mddocs/

Command Runner

A comprehensive platform-agnostic command execution library that serves as an enhanced replacement for subprocess.popen and subprocess.check_output. It provides robust handling of command timeouts, encoding issues, and platform differences between Windows and Linux/Unix systems, along with advanced features like live output capture, privilege elevation, and process management.

Package Information

  • Package Name: command_runner
  • Package Type: pypi
  • Language: Python
  • Installation: pip install command_runner
  • Python Compatibility: 2.7+
  • Platform Support: Windows, Linux, macOS, Unix

Core Imports

from command_runner import command_runner

For threaded execution (Python 3+ only):

from command_runner import command_runner_threaded

For privilege elevation:

from command_runner.elevate import elevate

Basic Usage

from command_runner import command_runner

# Simple command execution
exit_code, output = command_runner('ping 127.0.0.1', timeout=10)
print(f"Exit code: {exit_code}")
print(f"Output: {output}")

# Cross-platform command execution
if os.name == 'nt':
    cmd = 'ping 127.0.0.1 -n 4'
else:
    cmd = ['ping', '-c', '4', '127.0.0.1']

exit_code, output = command_runner(cmd, encoding='utf-8')

Architecture

Command Runner is built around a flexible execution framework:

  • Main Function: command_runner() provides the primary interface with extensive configuration options
  • Execution Methods: Two capture methods (monitor/poller) for different performance and feature trade-offs
  • Stream Handling: Sophisticated output redirection supporting files, queues, callbacks, and live display
  • Process Management: Complete process tree control with priority setting and reliable termination
  • Platform Abstraction: Unified interface handling Windows/Unix differences transparently
  • Error Handling: Comprehensive exception handling with consistent exit codes and partial output recovery

Capabilities

Core Command Execution

Primary command execution functionality with comprehensive timeout handling, encoding support, and output capture. Handles both string and list command formats across platforms with extensive configuration options.

def command_runner(
    command,                    # Union[str, List[str]] - Command to execute
    valid_exit_codes=False,     # Union[List[int], bool] - Accepted exit codes
    timeout=3600,               # Optional[int] - Timeout in seconds
    shell=False,                # bool - Use shell execution
    encoding=None,              # Optional[Union[str, bool]] - Output encoding
    stdin=None,                 # Optional[Union[int, str, Callable, queue.Queue]]
    stdout=None,                # Optional[Union[int, str, Callable, queue.Queue]]
    stderr=None,                # Optional[Union[int, str, Callable, queue.Queue]]
    no_close_queues=False,      # Optional[bool] - Keep queues open
    windows_no_window=False,    # bool - Hide window on Windows
    live_output=False,          # bool - Show output during execution
    method="monitor",           # str - Capture method ("monitor" or "poller")
    check_interval=0.05,        # float - Polling interval
    stop_on=None,               # Callable - Custom stop condition
    on_exit=None,               # Callable - Exit callback
    process_callback=None,      # Callable - Process info callback
    split_streams=False,        # bool - Return separate stdout/stderr
    silent=False,               # bool - Suppress logging
    priority=None,              # Union[int, str] - Process priority
    io_priority=None,           # str - IO priority
    heartbeat=0,                # int - Heartbeat logging interval
    **kwargs                    # Any - Additional subprocess arguments
):
    """
    Execute commands with advanced timeout, encoding, and output handling.
    
    Returns:
    Union[Tuple[int, Optional[Union[bytes, str]]], 
          Tuple[int, Optional[Union[bytes, str]], Optional[Union[bytes, str]]]]
    """
def command_runner_threaded(*args, **kwargs):
    """
    Threaded version returning concurrent.Future result (Python 3+ only).
    
    Returns:
    concurrent.futures.Future
    """

Core Command Execution

Process Management

Process priority control and reliable process tree termination functionality. Includes cross-platform priority setting and comprehensive child process management.

def set_priority(pid, priority):
    """
    Set process priority.
    
    Args:
        pid (int): Process ID
        priority (Union[int, str]): Priority level
    """

def set_io_priority(pid, priority):
    """
    Set IO priority.
    
    Args:
        pid (int): Process ID  
        priority (str): IO priority level
    """

def kill_childs_mod(pid=None, itself=False, soft_kill=False):
    """
    Kill all children of a process.
    
    Args:
        pid (int, optional): Target process ID
        itself (bool): Kill parent process too
        soft_kill (bool): Use soft termination
        
    Returns:
        bool: Success status
    """

Process Management

Privilege Elevation

UAC/sudo elevation functionality compatible with CPython, Nuitka, PyInstaller, and other Python packaging systems. Provides seamless privilege escalation across Windows and Unix platforms.

def elevate(callable_function, *args, **kwargs):
    """
    Elevate privileges and execute function with admin/root rights.
    
    Args:
        callable_function: Function to execute with elevation
        *args: Arguments to pass to function
        **kwargs: Keyword arguments to pass to function
    """

def is_admin():
    """
    Check if current process has administrative privileges.
    
    Returns:
        bool: True if admin privileges present
    """

Privilege Elevation

Utility Functions

Helper functions for encoding conversion, command preparation, and system compatibility. Includes constants for priority levels and platform-specific configurations.

def to_encoding(process_output, encoding, errors):
    """
    Convert bytes output to string with error handling.
    
    Args:
        process_output (Union[str, bytes]): Process output
        encoding (Optional[str]): Target encoding
        errors (str): Error handling strategy
        
    Returns:
        str: Converted string output
    """

def deferred_command(command, defer_time=300):
    """
    Launch detached command with delay.
    
    Args:
        command (str): Command to execute
        defer_time (int): Delay in seconds
    """

Utility Functions

Constants and Types

# Subprocess pipe reference
PIPE = subprocess.PIPE

# Priority level mappings (platform-specific)
PRIORITIES = {
    "process": {
        "verylow": ...,
        "low": ..., 
        "normal": ...,
        "high": ...,
        "rt": ...
    },
    "io": {
        "low": ...,
        "normal": ...,
        "high": ...
    }
}

# Special exit codes
# -250: Incompatible arguments
# -251: stop_on function returned True
# -252: KeyboardInterrupt  
# -253: FileNotFoundError, OSError, IOError
# -254: Timeout
# -255: Uncaught exceptions

Exception Classes

class TimeoutExpired(BaseException):
    """Command timeout exception."""
    def __init__(self, cmd, timeout, output=None, stderr=None): ...
    
class InterruptGetOutput(BaseException):
    """Base class for output capture on interruption."""
    def __init__(self, output): ...
    
class KbdInterruptGetOutput(InterruptGetOutput):
    """Keyboard interrupt with output capture."""
    
class StopOnInterrupt(InterruptGetOutput):
    """Stop condition interrupt with output capture."""