background_process_manager¶
Background Process Manager Module.
This module provides a utility for managing long-running background processes in a shell environment. It handles process spawning, monitoring, and termination, allowing for safer and more controlled execution of background tasks.
The module is particularly useful for managing persistent services, daemons, or any long-running commands that need to be started, monitored, and cleanly terminated programmatically.
Examples
>>> from haive.tools.toolkits.dev.shell.background_process_manager import BackgroundProcessManager
>>> bg_manager = BackgroundProcessManager()
>>> bg_manager.run_background("npm run dev")
{'success': True, 'pid': 12345, 'message': '✅ npm run dev is running in the background'}
>>> bg_manager.list_running()
[{'pid': 12345, 'command': 'npm run dev', 'status': 'running'}]
>>> bg_manager.stop_process(12345)
{'success': True, 'message': '✅ Process 12345 stopped'}
Classes¶
Manages long-running background processes. |
Module Contents¶
- class background_process_manager.BackgroundProcessManager¶
Manages long-running background processes.
This class provides methods to start, monitor, and stop background processes, as well as utilities to inspect network listeners and track resource usage.
- processes¶
Dictionary mapping process IDs to subprocess.Popen objects.
- cleanup() dict[str, Any]¶
Stop all managed processes and clean up resources.
- Returns:
success: Boolean indicating whether cleanup was successful.
stopped: Number of processes stopped.
errors: List of errors encountered during cleanup.
- Return type:
Dictionary containing
- get_process_logs(pid: int, max_lines: int = 100) dict[str, Any]¶
Get stdout and stderr output from a running process.
- Parameters:
pid – The process ID to get logs for.
max_lines – Maximum number of lines to return.
- Returns:
success: Boolean indicating whether logs were retrieved successfully.
stdout: Standard output from the process.
stderr: Standard error from the process.
error: Error details if logs could not be retrieved.
- Return type:
Dictionary containing
- list_network_listeners() list[dict[str, Any]]¶
List all processes listening on network ports.
- Returns:
port: The port number being listened on.
process: The process name.
pid: The process ID.
address: The binding address.
protocol: The protocol (TCP/UDP).
- Return type:
List of dictionaries containing information about each listening process
- list_running() list[dict[str, Any]]¶
List currently running background processes.
- Returns:
pid: Process ID.
command: The command that started the process.
status: Current status of the process.
memory: Memory usage in MB (if available).
cpu: CPU usage percentage (if available).
- Return type:
List of dictionaries containing information about each running process
- run_background(command: str, cwd: str | None = None, env: dict[str, str] | None = None) dict[str, Any]¶
Start a long-running command in the background.
- Parameters:
command – The shell command to run in the background.
cwd – Optional working directory for the process.
env – Optional environment variables for the process.
- Returns:
success: Boolean indicating whether the process started successfully.
pid: Process ID of the spawned process if successful.
message: Success or error message.
error: Error details if the process failed to start.
- Return type:
Dictionary containing
- Raises:
OSError – If there’s an issue with the system call to create the process.
- stop_process(pid: int, force: bool = False) dict[str, Any]¶
Stop a specific background process.
- Parameters:
pid – The process ID to stop.
force – Whether to forcefully kill the process (SIGKILL) rather than gracefully terminate it (SIGTERM).
- Returns:
success: Boolean indicating whether the process was stopped successfully.
message: Success or error message.
error: Error details if the process could not be stopped.
- Return type:
Dictionary containing
- processes¶