ProcessManager
ProcessManager executes structured CommandOperation objects and
low-level commands, with secure-by-default shell execution.
- class oduit.process_manager.ProcessManager[source]
Bases:
BaseProcessManagerCross-platform process execution manager for Odoo operations.
Provides comprehensive process management functionality including: - Command execution with output streaming and error handling - Cross-platform sudo authentication via stdin - Interactive shell support with pseudo-terminal (PTY) handling - Real-time output colorization and formatting - JSON-structured logging for programmatic consumption - Signal-safe process termination and cleanup
This class handles the complexities of running Odoo commands across different operating systems while providing consistent interfaces for both interactive and automated usage scenarios.
Example
>>> pm = ProcessManager() >>> result = pm.run_command(['echo', 'hello world']) >>> print(result['success']) # True >>> print(result['output']) # 'hello world\n'
>>> # Interactive shell >>> pm.run_interactive_shell(['bash'])
- run_operation(command_operation: CommandOperation, verbose: bool = False, suppress_output: bool = False) dict[str, Any][source]
Execute a CommandOperation directly.
For regular ProcessManager, this builds the command from the operation and executes it normally with enhanced result processing.
- Parameters:
command_operation – Structured command operation with metadata
verbose – Enable verbose output
suppress_output – Suppress output to console
- Returns:
Dict containing execution results
- run_command(cmd: list[str], stop_on_error: bool = False, compact: bool = False, verbose: bool = False, suppress_output: bool = False) dict[str, Any][source]
Execute a command with comprehensive output handling and error management.
This method provides a unified interface for running system commands with proper error handling, output streaming, and optional sudo support.
- Parameters:
cmd – Command to execute as list of strings (e.g., [‘ls’, ‘-la’])
stop_on_error – If True, terminate on first error encountered
compact – If True, use compact output formatting
verbose – If True, print the command being executed
suppress_output – If True, suppress all output to console
- Returns:
success (bool): True if command executed successfully
return_code (int): Process exit code (0 for success)
output (str): Combined stdout/stderr output
command (str): The executed command as string
error (str): Error message if execution failed
- Return type:
Dict containing execution results with keys
- Raises:
KeyboardInterrupt – Re-raised after proper process cleanup
Examples
>>> pm = ProcessManager() >>> result = pm.run_command(['echo', 'hello']) >>> print(result['success']) # True >>> print(result['output']) # 'hello\n'
>>> result = pm.run_command(['false']) # Command that fails >>> print(result['success']) # False >>> print(result['return_code']) # 1
- run_command_yielding(cmd: list[str], stop_on_error: bool = False, compact: bool = False, verbose: bool = False, suppress_output: bool = False) Generator[dict[str, Any], None, None][source]
Generator version of run_command that yields lines as they arrive
- Parameters:
cmd – Command to execute
stop_on_error – Stop execution on first error pattern
compact – Only show relevant lines (dots, errors, warnings)
verbose – Print command before execution
suppress_output – Don’t print lines to console (only yield them)
- Yields:
For each line –
dict: {‘line’: str, ‘formatted’: str, ‘should_show’: bool, ‘is_error’: bool, ‘process_running’: bool}
Final yield:
dict: {‘result’: dict, ‘process_running’: False}
- run_shell_command(cmd: list[str] | str, verbose: bool = False, capture_output: bool = False, allow_shell: bool = False, input_data: str | None = None) dict[str, Any][source]
Run a shell command that may receive piped input
- Parameters:
cmd – Either a list of command arguments or a string to be evaluated by shell
verbose – Print command before execution
capture_output – Capture stdout/stderr instead of inheriting
allow_shell – Allow shell string evaluation when cmd is a string
input_data – Optional data to send to stdin
Usage Examples
from oduit import ConfigLoader, ProcessManager
from oduit.builders import InstallCommandBuilder
loader = ConfigLoader()
config = loader.load_config("dev")
manager = ProcessManager()
operation = InstallCommandBuilder(config, "sale").build_operation()
result = manager.run_operation(operation, verbose=True)
shell_result = manager.run_shell_command(
["python", "-c", "print('hello')"],
capture_output=True,
)
Key Methods
run_operation(): execute a structured command with automatic result parsingrun_command(): run a direct command listrun_command_yielding(): stream structured output eventsrun_shell_command(): execute list commands safely, or string commands with explicitallow_shell=Trueclear_sudo_password(): clear cached sudo credentials
Notes
run_command()does not expose a publictimeout=...parameterstring shell commands require
allow_shell=Trueembedded execution is not a supported public
ProcessManagermode