ProcessManager

ProcessManager executes structured CommandOperation objects and low-level commands, with secure-by-default shell execution.

class oduit.process_manager.ProcessManager[source]

Bases: BaseProcessManager

Cross-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.

_sudo_password

Cached sudo password for repeated operations

Type:

str | None

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'])
__init__() None[source]
clear_sudo_password() None[source]

Clear cached sudo password from memory.

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

static run_interactive_shell(cmd: list[str]) int[source]

Run an interactive shell command with proper cross-platform handling

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 parsing

  • run_command(): run a direct command list

  • run_command_yielding(): stream structured output events

  • run_shell_command(): execute list commands safely, or string commands with explicit allow_shell=True

  • clear_sudo_password(): clear cached sudo credentials

Notes

  • run_command() does not expose a public timeout=... parameter

  • string shell commands require allow_shell=True

  • embedded execution is not a supported public ProcessManager mode